From 5d8ca525a41057ef1934d0377ad97b7bedca0c32 Mon Sep 17 00:00:00 2001 From: Urban Ishimwe Date: Tue, 13 Oct 2020 07:36:16 +0200 Subject: [PATCH] deprecate output-http non-compatible clients (#833) for easy readability check changes by their underlying commits! these benchmarks address the whole operation of the request cycle in goreplay. _**goos: linux goarch: amd64**_ **Using Compatible client:** ``` BenchmarkHTTPOutput-4 10417 118969 ns/op 12172 B/op 93 allocs/op BenchmarkHTTPOutputTLS-4 9136 132929 ns/op 12448 B/op 97 allocs/op ``` **Using non-compatible client** ``` BenchmarkHTTPOutput-4 859 1175040 ns/op 15598 B/op 46 allocs/op BenchmarkHTTPOutputTLS-4 880 1189643 ns/op 15544 B/op 52 allocs/op ``` Binary size reduced: **7%** from these benchmarks, we may trade allocations with performance and memory! --- http_client.go | 530 ------------------------------------------- http_client_test.go | 532 -------------------------------------------- http_prettifier.go | 37 ++- input_kafka.go | 5 +- input_raw_test.go | 43 ++-- kafka.go | 12 +- output_http.go | 392 ++++++++++++++++---------------- output_http_test.go | 48 +++- output_kafka.go | 28 +-- proto/fuzz.go | 4 +- proto/proto.go | 20 +- proto/proto_test.go | 38 ++-- protocol.go | 12 +- settings.go | 10 +- 14 files changed, 308 insertions(+), 1403 deletions(-) delete mode 100644 http_client.go delete mode 100644 http_client_test.go diff --git a/http_client.go b/http_client.go deleted file mode 100644 index 92c6065..0000000 --- a/http_client.go +++ /dev/null @@ -1,530 +0,0 @@ -package main - -import ( - "bufio" - "bytes" - "crypto/tls" - "encoding/base64" - "io" - "log" - "net" - "net/http" - "net/http/httputil" - "net/url" - "os" - "runtime/debug" - "strconv" - "strings" - "sync" - "syscall" - "time" - - "github.com/buger/goreplay/proto" -) - -var httpMu sync.Mutex - -const ( - readChunkSize = 64 * 1024 - maxResponseSize = 1073741824 -) - -var chunkedSuffix = []byte("0\r\n\r\n") - -var defaultPorts = map[string]string{ - "http": "80", - "https": "443", -} - -type HTTPClientConfig struct { - FollowRedirects int - Debug bool - OriginalHost bool - ConnectionTimeout time.Duration - Timeout time.Duration - ResponseBufferSize int - CompatibilityMode bool -} - -type HTTPClient struct { - baseURL string - scheme string - host string - auth string - conn net.Conn - proxy *url.URL - proxyAuth string - respBuf []byte - config *HTTPClientConfig - goClient *http.Client - redirectsCount int -} - -func NewHTTPClient(baseURL string, config *HTTPClientConfig) *HTTPClient { - if !strings.HasPrefix(baseURL, "http") { - baseURL = "http://" + baseURL - } - - u, _ := url.Parse(baseURL) - - if config.Timeout == 0 { - config.Timeout = time.Second - } - - config.ConnectionTimeout = config.Timeout - - if config.ResponseBufferSize == 0 { - config.ResponseBufferSize = 100 * 1024 // 100kb - } - - client := new(HTTPClient) - client.baseURL = u.String() - client.host = u.Host - client.scheme = u.Scheme - client.respBuf = make([]byte, config.ResponseBufferSize) - client.config = config - - if config.CompatibilityMode { - client.goClient = &http.Client{ - // #TODO - // CheckRedirect: redirectPolicyFunc, - } - } - - if u.User != nil { - client.auth = "Basic " + base64.StdEncoding.EncodeToString([]byte(u.User.String())) - } - - client.proxy, _ = http.ProxyFromEnvironment(&http.Request{URL: u}) - - if client.isProxy() && client.proxy.User != nil { - client.proxyAuth = "Basic " + base64.StdEncoding.EncodeToString([]byte(client.proxy.User.String())) - } - - return client -} - -func (c *HTTPClient) Connect() (err error) { - c.Disconnect() - - var toDial string - if !strings.Contains(c.host, ":") { - toDial = c.host + ":" + defaultPorts[c.scheme] - } else { - toDial = c.host - } - - if c.isProxy() { - if c.proxy.Scheme != "http" { - panic("Unsupported HTTP Proxy method") - } - Debug(3, "[HTTPClient] Connecting to proxy", c.proxy.String(), "<>", toDial) - c.conn, err = net.DialTimeout("tcp", c.proxy.Host, c.config.ConnectionTimeout) - if err != nil { - return - } - if c.scheme == "https" { - c.conn.Write([]byte("CONNECT " + toDial + " HTTP/1.1\r\n")) - if c.proxyAuth != "" { - c.conn.Write([]byte("Proxy-Authorization: " + c.proxyAuth + "\r\n")) - } - c.conn.Write([]byte("\r\n")) - br := bufio.NewReader(c.conn) - l, _, err := br.ReadLine() - if err != nil { - return err - } - if len(l) < 12 { - panic("HTTP proxy did not respond correctly") - } - status := l[9:12] - if !bytes.Equal(status, []byte("200")) { - panic("HTTP proxy did not respond correctly") - } - for { - // Read until we find the empty line - l, _, err := br.ReadLine() - if err != nil { - return err - } - if len(l) == 0 { - break - } - } - } - Debug(3, "[HTTPClient] Proxy successfully connected") - } else { - c.conn, err = net.DialTimeout("tcp", toDial, c.config.ConnectionTimeout) - if err != nil { - return - } - } - - if c.scheme == "https" { - // Wrap our socket in TLS - Debug(3, "[HTTPClient] Wrapping socket in TLS", c.host) - tlsConn := tls.Client(c.conn, &tls.Config{InsecureSkipVerify: true, ServerName: c.host}) - - if err = tlsConn.Handshake(); err != nil { - return - } - - c.conn = tlsConn - Debug(3, "[HTTPClient] Successfully wrapped in TLS") - } - - return -} - -func (c *HTTPClient) Disconnect() { - if c.conn != nil { - c.conn.Close() - c.conn = nil - Debug(3, "[HTTP] Disconnected: ", c.baseURL) - } -} - -func isSyscallOpError(err error, errno syscall.Errno) bool { - if opErr, ok := err.(*net.OpError); ok { - if syscallErr, ok := opErr.Err.(*os.SyscallError); ok { - return syscallErr.Err == errno - } - } - return err == errno -} - -func (c *HTTPClient) isAlive(readBytes *int) bool { - // Ready 1 byte from socket without timeout to check if it not closed - c.conn.SetReadDeadline(time.Now().Add(time.Millisecond)) - n, err := c.conn.Read(c.respBuf[:1]) - - if err == io.EOF || isSyscallOpError(err, syscall.ECONNRESET) { - Debug(3, "[HTTPClient] connection closed, reconnecting") - return false - } - if isSyscallOpError(err, syscall.EPIPE) { - Debug(3, "[HTTPClient] Detected broken pipe.", err) - return false - } - if n != 0 { - *readBytes += n - Debug(3, "[HTTPClient] isAlive readBytes ", *readBytes) - } - return true -} - -func (c *HTTPClient) SendGoClient(data []byte) ([]byte, error) { - var req *http.Request - var resp *http.Response - var err error - - req, err = http.ReadRequest(bufio.NewReader(bytes.NewBuffer(data))) - if err != nil { - return nil, err - } - - if !c.config.OriginalHost { - req.Host = c.host - } - - if c.auth != "" { - req.Header.Add("Authorization", c.auth) - } - - req.URL, _ = url.ParseRequestURI(c.scheme + "://" + c.host + req.RequestURI) - req.RequestURI = "" - - resp, err = c.goClient.Do(req) - if err != nil { - return nil, err - } - - return httputil.DumpResponse(resp, true) -} - -func (c *HTTPClient) Send(data []byte) (response []byte, err error) { - // Don't exit on panic - defer func() { - if r := recover(); r != nil { - Debug(3, "[HTTPClient]", r, string(data)) - - if _, ok := r.(error); ok { - log.Println("[HTTPClient] Failed to send request: ", string(data)) - log.Println("[HTTPClient] Response: ", string(response)) - log.Println("PANIC: pkg:", r, string(debug.Stack())) - } - } - }() - - if c.config.CompatibilityMode { - return c.SendGoClient(data) - } - - var readBytes int - if c.conn == nil || !c.isAlive(&readBytes) { - Debug(3, "[HTTPClient] Connecting:", c.baseURL) - if err = c.Connect(); err != nil { - Debug(1, "[HTTPClient] Connection error:", err) - response = errorPayload(HTTP_CONNECTION_ERROR) - return - } - } - - timeout := time.Now().Add(c.config.Timeout) - - c.conn.SetWriteDeadline(timeout) - - if !c.config.OriginalHost { - data = proto.SetHost(data, []byte(c.baseURL), []byte(c.host)) - } - - if c.isProxy() && c.scheme == "http" { - path := proto.Path(data) - if len(path) > 0 && path[0] == '/' { - data = proto.SetPath(data, c.proxyPath(path)) - if c.proxyAuth != "" { - data = proto.SetHeader(data, []byte("Proxy-Authorization"), []byte(c.proxyAuth)) - } - } - } - - if c.auth != "" { - data = proto.SetHeader(data, []byte("Authorization"), []byte(c.auth)) - } - - if c.config.Debug { - Debug(3, "[HTTPClient] Sending:", string(data)) - } - - return c.send(data, readBytes, timeout) -} - -func (c *HTTPClient) send(data []byte, readBytes int, timeout time.Time) (response []byte, err error) { - var payload []byte - var n int - if _, err = c.conn.Write(data); err != nil { - Debug(1, "[HTTPClient] Write error:", err, c.baseURL) - response = errorPayload(HTTP_TIMEOUT) - c.Disconnect() - return - } - - var currentChunk []byte - timeout = time.Now().Add(c.config.Timeout) - chunked := false - contentLength := -1 - currentContentLength := 0 - chunks := 0 - - for { - c.conn.SetReadDeadline(timeout) - - if readBytes < len(c.respBuf) { - n, err = c.conn.Read(c.respBuf[readBytes:]) - readBytes += n - chunks++ - - // First chunk - if chunked || contentLength != -1 { - currentContentLength += n - } else { - // If headers are finished - var firstEmptyLine = bytes.Index(c.respBuf[:readBytes], proto.EmptyLine) - if firstEmptyLine != -1 { - if bytes.Equal(proto.Header(c.respBuf[:readBytes], []byte("Transfer-Encoding")), []byte("chunked")) { - chunked = true - } else { - status, _ := strconv.Atoi(string(proto.Status(c.respBuf[:readBytes]))) - // We want to soak up all 100 Continues received to get the real result code - if status >= 100 && status < 200 { - timeout = time.Now().Add(c.config.Timeout) - var deleteLen = firstEmptyLine + len(proto.EmptyLine) - copy(c.respBuf, c.respBuf[deleteLen:readBytes]) - readBytes -= deleteLen - chunks-- - continue - } else if status == 204 || status == 304 { - contentLength = 0 - break - } else { - l := proto.Header(c.respBuf[:readBytes], []byte("Content-Length")) - if len(l) > 0 { - contentLength, _ = strconv.Atoi(string(l)) - } - } - } - - currentContentLength += len(proto.Body(c.respBuf[:readBytes])) - } - } - - if chunked { - // Check if chunked message finished - if bytes.HasSuffix(c.respBuf[:readBytes], chunkedSuffix) { - break - } - } else if contentLength != -1 { - if currentContentLength > contentLength { - Debug(3, "[HTTPClient] disconnected, wrong length", currentContentLength, contentLength) - c.Disconnect() - break - } else if currentContentLength == contentLength { - break - } - } - - if err != nil { - if err == io.EOF { - err = nil - } - break - } - } else { - if currentChunk == nil { - currentChunk = make([]byte, readChunkSize) - } - - n, err = c.conn.Read(currentChunk) - - readBytes += int(n) - chunks++ - currentContentLength += n - - if chunked { - // Check if chunked message finished - if bytes.HasSuffix(currentChunk[:n], chunkedSuffix) { - break - } - } else if contentLength != -1 { - if currentContentLength > contentLength { - Debug(3, "[HTTPClient] disconnected, wrong length", currentContentLength, contentLength) - c.Disconnect() - break - } else if currentContentLength == contentLength { - break - } - } else { - Debug(3, "[HTTPClient] disconnected, can't find Content-Length or Chunked") - c.Disconnect() - break - } - - if err == io.EOF { - break - } else if err != nil { - Debug(3, "[HTTPClient] Read the whole body error:", err, c.baseURL) - break - } - - } - - if readBytes >= maxResponseSize { - Debug(3, "[HTTPClient] Body is more than the max size", maxResponseSize, - c.baseURL) - break - } - - // For following chunks expect less timeout - timeout = time.Now().Add(c.config.Timeout / 5) - } - - if err != nil && readBytes == 0 { - Debug(3, "[HTTPClient] Response read timeout error", err, c.conn, readBytes, string(c.respBuf[:readBytes])) - response = errorPayload(HTTP_TIMEOUT) - c.Disconnect() - return - } - - if readBytes < 4 || string(c.respBuf[:4]) != "HTTP" { - maxRead := 100 - if readBytes < maxRead { - maxRead = readBytes - } - Debug(3, "[HTTPClient] Response read unknown error", err, c.conn, readBytes, string(c.respBuf[:maxRead])) - response = errorPayload(HTTP_UNKNOWN_ERROR) - c.Disconnect() - return - } - - if readBytes > len(c.respBuf) { - readBytes = len(c.respBuf) - } - payload = make([]byte, readBytes) - copy(payload, c.respBuf[:readBytes]) - - if c.config.Debug { - Debug(3, "[HTTPClient] Received:", string(payload)) - } - - if c.config.FollowRedirects > 0 && c.redirectsCount < c.config.FollowRedirects { - status := payload[9:12] - - // 3xx requests - if status[0] == '3' { - c.redirectsCount++ - - location := proto.Header(payload, []byte("Location")) - redirectPayload := proto.SetPath(data, location) - - Debug(3, "[HTTPClient] Redirecting to: "+string(location)) - - return c.Send(redirectPayload) - } - } - - if bytes.Equal(proto.Status(payload), []byte("400")) { - Debug(3, "[HTTPClient] Closed connection on 400 response") - c.Disconnect() - } - - c.redirectsCount = 0 - - return payload, err -} - -func (c *HTTPClient) Get(path string) (response []byte, err error) { - payload := "GET " + path + " HTTP/1.1\r\n\r\n" - - return c.Send([]byte(payload)) -} - -func (c *HTTPClient) Post(path string, body []byte) (response []byte, err error) { - payload := "POST " + path + " HTTP/1.1\r\n" - payload += "Content-Length: " + strconv.Itoa(len(body)) + "\r\n\r\n" - payload += string(body) - - return c.Send([]byte(payload)) -} - -func (c *HTTPClient) proxyPath(path []byte) []byte { - return append([]byte(c.scheme+"://"+c.host), path...) -} - -func (c *HTTPClient) isProxy() bool { - return c.proxy != nil -} - -const ( - // https://support.cloudflare.com/hc/en-us/articles/200171936-Error-520-Web-server-is-returning-an-unknown-error - HTTP_UNKNOWN_ERROR = "520" - // https://support.cloudflare.com/hc/en-us/articles/200171916-Error-521-Web-server-is-down - HTTP_CONNECTION_ERROR = "521" - // https://support.cloudflare.com/hc/en-us/articles/200171906-Error-522-Connection-timed-out - HTTP_CONNECTION_TIMEOUT = "522" - // https://support.cloudflare.com/hc/en-us/articles/200171946-Error-523-Origin-is-unreachable - HTTP_UNREACHABLE = "523" - // https://support.cloudflare.com/hc/en-us/articles/200171926-Error-524-A-timeout-occurred - HTTP_TIMEOUT = "524" -) - -var errorPayloadTemplate = "HTTP/1.1 202 Accepted\r\nDate: Mon, 17 Aug 2015 14:10:11 GMT\r\nContent-Length: 0\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n" - -func errorPayload(errorCode string) []byte { - payload := make([]byte, len(errorPayloadTemplate)) - copy(payload, errorPayloadTemplate) - - copy(payload[29:58], []byte(time.Now().Format(time.RFC1123))) - copy(payload[9:12], errorCode) - - return payload -} diff --git a/http_client_test.go b/http_client_test.go deleted file mode 100644 index 0ad67fc..0000000 --- a/http_client_test.go +++ /dev/null @@ -1,532 +0,0 @@ -package main - -import ( - "bytes" - "crypto/rand" - "io/ioutil" - _ "log" - "net" - "net/http" - "net/http/httptest" - "net/http/httputil" - _ "reflect" - "strings" - "sync" - "testing" - "time" - - "github.com/buger/goreplay/proto" -) - -func TestHTTPClientURLPort(t *testing.T) { - c1 := NewHTTPClient("http://example.com", &HTTPClientConfig{}) - if c1.baseURL != "http://example.com" { - t.Error("Sould not add 80 port for http:", c1.baseURL) - } - - c2 := NewHTTPClient("https://example.com", &HTTPClientConfig{}) - if c2.baseURL != "https://example.com" { - t.Error("Sould not add 443 port for https:", c2.baseURL) - } - - c3 := NewHTTPClient("https://example.com:1", &HTTPClientConfig{}) - if c3.baseURL != "https://example.com:1" { - t.Error("Sould use specified port:", c3.baseURL) - } - - c4 := NewHTTPClient("example.com", &HTTPClientConfig{}) - if c4.baseURL != "http://example.com" { - t.Error("Sould not add default protocol:", c4.baseURL) - } -} - -func TestHTTPClientSend(t *testing.T) { - wg := new(sync.WaitGroup) - - 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") - } - - return []byte("") - } - - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - - if r.Method == "POST" { - defer r.Body.Close() - body, _ := ioutil.ReadAll(r.Body) - - if len(r.TransferEncoding) > 0 && r.TransferEncoding[0] == "chunked" { - if string(body) != "Wikipedia in\r\n\r\nchunks." { - t.Error("Wrong POST body:", body, string(body)) - } - } else { - if string(body) != "a=1&b=2" { - buf, _ := httputil.DumpRequest(r, true) - t.Error("Wrong POST body:", string(body), string(buf)) - } - } - } - - wg.Done() - })) - defer server.Close() - - client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true}) - - wg.Add(4) - client.Send(payload("POST")) - client.Send(payload("GET")) - client.Send(payload("POST_CHUNKED")) - client.Send(payload("POST")) - - wg.Wait() - - client = NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true, CompatibilityMode: true}) - - wg.Add(4) - - if _, err := client.Send(payload("POST")); err != nil { - t.Fatal(err.Error()) - } - if _, err := client.Send(payload("GET")); err != nil { - t.Fatal(err.Error()) - } - if _, err := client.Send(payload("POST_CHUNKED")); err != nil { - t.Fatal(err.Error()) - } - if _, err := client.Send(payload("POST")); err != nil { - t.Fatal(err.Error()) - } - - wg.Wait() -} - -func TestHTTPClientResonseByClose(t *testing.T) { - wg := new(sync.WaitGroup) - - payload := []byte("GET / HTTP/1.1\r\n\r\n") - ln, _ := net.Listen("tcp", ":0") - go func() { - for { - conn, _ := ln.Accept() - buf := make([]byte, 4096) - conn.Read(buf) - - conn.Write([]byte("HTTP/1.1 200 OK\r\n\r\n")) - conn.Write([]byte("ab")) - conn.Close() - - wg.Done() - } - }() - - client := NewHTTPClient(ln.Addr().String(), &HTTPClientConfig{Debug: true}) - - wg.Add(1) - resp, _ := client.Send(payload) - - if !bytes.Equal(resp, []byte("HTTP/1.1 200 OK\r\n\r\nab")) { - t.Error("Should return valid response", string(resp)) - } - - wg.Wait() -} - -// https://github.com/buger/gor/issues/184 -func TestHTTPClientResponseBuffer(t *testing.T) { - testCases := []struct { - name string - responseSize int - buffserSize int - expectedSize int - timeout time.Duration - }{ - {"Chunked, buffer overflow", 10 * 1024, 1024, 1024, 50 * time.Millisecond}, - - {"Chunked, fits buffer", 10 * 1024, 64 * 1024, 10*1024 + 145 /* headers length + chunked meta */, 50 * time.Millisecond}, - - {"Content-Length, buffer overflow", 1024, 1000, 1000, 50 * time.Millisecond}, - - {"Content-Length, fits buffer", 1024, 64 * 1024, 1024 + 118, 50 * time.Millisecond}, - } - - for _, tc := range testCases { - wg := new(sync.WaitGroup) - - payload := []byte("GET / HTTP/1.1\r\n\r\n") - - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - - size := tc.responseSize // 1kb - rb := make([]byte, size) - rand.Read(rb) - - w.Write(rb[:size/2]) - w.Write(rb[size/2:]) - - wg.Done() - })) - - client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true, ResponseBufferSize: tc.buffserSize, Timeout: 100 * time.Millisecond}) - - wg.Add(2) - - start := time.Now() - client.Send(payload) - resp, err := client.Send(payload) - stop := time.Now() - - if err != nil { - t.Error("Request error", err) - } - - if stop.Sub(start) > tc.timeout { - t.Error("Request took too long", stop.Sub(start), tc.timeout) - } - - if len(resp) != tc.expectedSize { - t.Error(tc.name, " - Wrong response size:", tc.expectedSize, len(resp)) - } else { - if !bytes.Equal(resp[0:8], []byte("HTTP/1.1")) { - t.Error(tc.name, " - Response buffer contains data from previous request", string(resp), len(resp)) - } - } - - wg.Wait() - server.Close() - } -} - -func TestHTTPClientHTTPSSend(t *testing.T) { - wg := new(sync.WaitGroup) - - 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") - } - - return []byte("") - } - - server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - - if r.Method == "POST" { - defer r.Body.Close() - body, _ := ioutil.ReadAll(r.Body) - - if len(r.TransferEncoding) > 0 && r.TransferEncoding[0] == "chunked" { - if string(body) != "Wikipedia in\r\n\r\nchunks." { - t.Error("Wrong POST body:", body, string(body)) - } - } else { - if string(body) != "a=1&b=2" { - buf, _ := httputil.DumpRequest(r, true) - t.Error("Wrong POST body:", string(body), string(buf)) - } - } - } - - wg.Done() - })) - defer server.Close() - - client := NewHTTPClient(server.URL, &HTTPClientConfig{}) - - wg.Add(4) - client.Send(payload("POST")) - client.Send(payload("GET")) - client.Send(payload("POST_CHUNKED")) - client.Send(payload("POST")) - - wg.Wait() -} - -func TestHTTPClientServerInstantDisconnect(t *testing.T) { - wg := new(sync.WaitGroup) - - GETPayload := []byte("GET / HTTP/1.1\r\n\r\n") - - ln, _ := net.Listen("tcp", "127.0.0.1:0") - defer ln.Close() - - go func() { - for { - conn, err := ln.Accept() - - if err != nil { - break - } - conn.Close() - wg.Done() - } - }() - - client := NewHTTPClient(ln.Addr().String(), &HTTPClientConfig{}) - - wg.Add(2) - client.Send(GETPayload) - client.Send(GETPayload) - - wg.Wait() -} - -func TestHTTPClientServerNoKeepAlive(t *testing.T) { - wg := new(sync.WaitGroup) - - GETPayload := []byte("GET / HTTP/1.1\r\n\r\n") - - ln, _ := net.Listen("tcp", "127.0.0.1:0") - defer ln.Close() - - go func() { - for { - conn, err := ln.Accept() - if err != nil { - break - } - - buf := make([]byte, 4096) - _, err = conn.Read(buf) - if err != nil { - t.Error("Error reading:", err.Error()) - } - conn.Write([]byte("OK")) - - // No keep-alive connections - conn.Close() - - wg.Done() - } - }() - - client := NewHTTPClient(ln.Addr().String(), &HTTPClientConfig{}) - - wg.Add(2) - client.Send(GETPayload) - client.Send(GETPayload) - - wg.Wait() -} - -func TestHTTPClientRedirect(t *testing.T) { - wg := new(sync.WaitGroup) - - GETPayload := []byte("GET / HTTP/1.1\r\n\r\n") - - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - - if r.URL.Path == "/" { - http.Redirect(w, r, "/new", 301) - } - - wg.Done() - })) - defer server.Close() - - client := NewHTTPClient(server.URL, &HTTPClientConfig{FollowRedirects: 1, Debug: false}) - - // Should do 2 queries - wg.Add(2) - client.Send(GETPayload) - - wg.Wait() -} - -func TestHTTPClientRedirectLimit(t *testing.T) { - wg := new(sync.WaitGroup) - - GETPayload := []byte("GET / HTTP/1.1\r\n\r\n") - - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - - if r.URL.Path == "/" { - http.Redirect(w, r, "/r1", 301) - } - - if r.URL.Path == "/r1" { - http.Redirect(w, r, "/r2", 301) - } - - if r.URL.Path == "/r2" { - http.Redirect(w, r, "/new", 301) - } - - wg.Done() - })) - defer server.Close() - - client := NewHTTPClient(server.URL, &HTTPClientConfig{FollowRedirects: 2, Debug: false}) - - // Have 3 redirects + 1 GET, but should do only 2 redirects + GET - wg.Add(3) - client.Send(GETPayload) - - wg.Wait() -} - -func TestHTTPClientKeepHeadersRedirect(t *testing.T) { - wg := new(sync.WaitGroup) - - GETPayload := []byte("GET / HTTP/1.1\r\n\r\n") - - GETPayload = proto.AddHeader(GETPayload, []byte("keep-header"), []byte("true")) - - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - - if r.URL.Path == "/" { - http.Redirect(w, r, "/new", 301) - } - - if r.Header.Get("keep-header") != "true" { - t.Errorf("Header keep-header was incorrect, got: %s, want: %s.", r.Header.Get("keep-header"), "true") - } - - wg.Done() - })) - defer server.Close() - - client := NewHTTPClient(server.URL, &HTTPClientConfig{FollowRedirects: 1, Debug: false}) - - wg.Add(2) - client.Send(GETPayload) - - wg.Wait() -} - -func TestHTTPClientBasicAuth(t *testing.T) { - wg := new(sync.WaitGroup) - wg.Add(2) - - GETPayload := []byte("GET / HTTP/1.1\r\n\r\n") - - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - user, pass, _ := r.BasicAuth() - - if user != "user" || pass != "pass" { - http.Error(w, "Unauthorized.", 401) - wg.Done() - return - } - - wg.Done() - })) - defer server.Close() - - client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: false}) - resp, _ := client.Send(GETPayload) - client.Disconnect() - - if !bytes.Equal(proto.Status(resp), []byte("401")) { - t.Error("Should return unauthorized error", string(resp)) - } - - authUrl := strings.Replace(server.URL, "http://", "http://user:pass@", -1) - client = NewHTTPClient(authUrl, &HTTPClientConfig{Debug: false}) - resp, _ = client.Send(GETPayload) - client.Disconnect() - - if !bytes.Equal(proto.Status(resp), []byte("200")) { - t.Error("Should return proper response", string(resp)) - } - - wg.Wait() -} - -func TestHTTPClientHandleHTTP10(t *testing.T) { - wg := new(sync.WaitGroup) - - GETPayload := []byte("GET http://foobar.com/path HTTP/1.0\r\n\r\n") - - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - - if r.URL.Path != "/path" { - t.Error("Path not match:", r.URL.Path) - } - - wg.Done() - })) - defer server.Close() - - client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true}) - - wg.Add(1) - client.Send(GETPayload) - - wg.Wait() -} - -// func TestHTTPClientErrors(t *testing.T) { -// req := []byte("GET http://foobar.com/path HTTP/1.0\r\n\r\n") - -// // Port not exists -// client := NewHTTPClient("http://127.0.0.1:1", &HTTPClientConfig{Debug: true}) -// if resp, err := client.Send(req); err != nil { -// if s := proto.Status(resp); !bytes.Equal(s, []byte("521")) { -// t.Error("Should return status 521 for connection refused, instead:", string(s), err) -// } -// } else { -// t.Error("Should throw error") -// } - -// client = NewHTTPClient("http://not.existing", &HTTPClientConfig{Debug: true}) -// if resp, err := client.Send(req); err != nil { -// if s := proto.Status(resp); !bytes.Equal(s, []byte("521")) { -// t.Error("Should return status 521 for no such host, instead:", string(s)) -// } -// } else { -// t.Error("Should throw error") -// } - -// // Non routable IP address to simulate connection timeout -// client = NewHTTPClient("http://10.255.255.1", &HTTPClientConfig{Debug: true, ConnectionTimeout: 100 * time.Millisecond}) - -// if resp, err := client.Send(req); err != nil { -// if s := proto.Status(resp); !bytes.Equal(s, []byte("521")) { -// t.Error("Should return status 521 for io/timeout:", string(s)) -// } -// } else { -// t.Error("Should throw error") -// } - -// // Connecting but io timeout on read -// ln, _ := net.Listen("tcp", "127.0.0.1:0") -// client = NewHTTPClient("http://"+ln.Addr().String(), &HTTPClientConfig{Debug: true, Timeout: 10 * time.Millisecond}) -// defer ln.Close() - -// if resp, err := client.Send(req); err != nil { -// if s := proto.Status(resp); !bytes.Equal(s, []byte("524")) { -// t.Error("Should return status 524 for io read, instead:", string(s)) -// } -// } else { -// t.Error("Should throw error") -// } - -// // Response read error read tcp [::1]:51128: connection reset by peer &{{0xc20802a000}} -// ln1, _ := net.Listen("tcp", "127.0.0.1:0") -// go func() { -// ln1.Accept() -// }() -// defer ln1.Close() - -// client = NewHTTPClient("http://"+ln1.Addr().String(), &HTTPClientConfig{Debug: true, Timeout: 10 * time.Millisecond}) - -// if resp, err := client.Send(req); err != nil { -// if s := proto.Status(resp); !bytes.Equal(s, []byte("524")) { -// t.Error("Should return status 524 for connection reset by peer, instead:", string(s)) -// } -// } else { -// t.Error("Should throw error") -// } -// } diff --git a/http_prettifier.go b/http_prettifier.go index 688e9bc..7a9169f 100644 --- a/http_prettifier.go +++ b/http_prettifier.go @@ -3,6 +3,7 @@ package main import ( "bytes" "compress/gzip" + "fmt" "io/ioutil" "net/http/httputil" "strconv" @@ -15,6 +16,13 @@ func prettifyHTTP(p []byte) []byte { head := p[:headSize] body := p[headSize:] + tEnc := bytes.Equal(proto.Header(body, []byte("Transfer-Encoding")), []byte("chunked")) + cEnc := bytes.Equal(proto.Header(body, []byte("Content-Encoding")), []byte("gzip")) + + if !(tEnc || cEnc) { + return p + } + headersPos := proto.MIMEHeadersEndPos(body) if headersPos < 5 || headersPos > len(body) { @@ -24,23 +32,8 @@ func prettifyHTTP(p []byte) []byte { headers := body[:headersPos] content := body[headersPos:] - var tEnc, cEnc []byte - proto.ParseHeaders([][]byte{headers}, func(header, value []byte) { - if bytes.EqualFold(header, []byte("Transfer-Encoding")) { - tEnc = value - } - - if bytes.EqualFold(header, []byte("Content-Encoding")) { - cEnc = value - } - }) - - if len(tEnc) == 0 && len(cEnc) == 0 { - return p - } - - if bytes.Equal(tEnc, []byte("chunked")) { - buf := bytes.NewBuffer(content) + if tEnc { + buf := bytes.NewReader(content) r := httputil.NewChunkedReader(buf) content, _ = ioutil.ReadAll(r) @@ -50,8 +43,8 @@ func prettifyHTTP(p []byte) []byte { headers = proto.SetHeader(headers, []byte("Content-Length"), []byte(newLen)) } - if bytes.Equal(cEnc, []byte("gzip")) { - buf := bytes.NewBuffer(content) + if cEnc { + buf := bytes.NewReader(content) g, err := gzip.NewReader(buf) if err != nil { @@ -59,7 +52,11 @@ func prettifyHTTP(p []byte) []byte { return []byte{} } - content, _ = ioutil.ReadAll(g) + content, err = ioutil.ReadAll(g) + if err != nil { + Debug(1, fmt.Sprintf("[HTTP-PRETTIFIER] %q", err)) + return p + } headers = proto.DeleteHeader(headers, []byte("Content-Encoding")) diff --git a/input_kafka.go b/input_kafka.go index 80dd2db..d168b7a 100644 --- a/input_kafka.go +++ b/input_kafka.go @@ -32,7 +32,6 @@ func NewKafkaInputWithTLS(address string, config *InputKafkaConfig, tlsConfig *K con = config.consumer } else { var err error - //con, err = sarama.NewConsumer([]string{config.Host}, c) con, err = sarama.NewConsumer(strings.Split(config.Host, ","), c) if err != nil { @@ -97,9 +96,9 @@ func (i *KafkaInput) Read(data []byte) (int, error) { return 0, err } - copy(data, buf) + n := copy(data, buf) - return len(buf), nil + return n, nil } diff --git a/input_raw_test.go b/input_raw_test.go index eb4eddf..a41bf2a 100644 --- a/input_raw_test.go +++ b/input_raw_test.go @@ -70,14 +70,13 @@ func TestRAWInputIPv4(t *testing.T) { } plugins.All = append(plugins.All, input, output) - client := NewHTTPClient("127.0.0.1:"+port, &HTTPClientConfig{}) - + addr := "http://127.0.0.1:" + port emitter := NewEmitter(quit) defer emitter.Close() go emitter.Start(plugins, Settings.Middleware) for i := 0; i < 10; i++ { wg.Add(2) - _, err = client.Get("/") + _, err = http.Get(addr) if err != nil { t.Error(err) return @@ -133,7 +132,7 @@ func TestRAWInputNoKeepAlive(t *testing.T) { } plugins.All = append(plugins.All, input, output) - client := NewHTTPClient("127.0.0.1:"+port, &HTTPClientConfig{}) + addr := "http://127.0.0.1:" + port emitter := NewEmitter(quit) go emitter.Start(plugins, Settings.Middleware) @@ -141,7 +140,7 @@ func TestRAWInputNoKeepAlive(t *testing.T) { for i := 0; i < 10; i++ { // request + response wg.Add(2) - _, err = client.Get("/") + _, err = http.Get(addr) if err != nil { t.Error(err) return @@ -198,14 +197,13 @@ func TestRAWInputIPv6(t *testing.T) { Outputs: []io.Writer{output}, } - client := NewHTTPClient(originAddr, &HTTPClientConfig{}) - emitter := NewEmitter(quit) + addr := "http://" + originAddr go emitter.Start(plugins, Settings.Middleware) for i := 0; i < 10; i++ { // request + response wg.Add(2) - _, err = client.Get("/") + _, err = http.Get(addr) if err != nil { t.Error(err) return @@ -256,7 +254,7 @@ func TestInputRAWChunkedEncoding(t *testing.T) { })) defer replay.Close() - httpOutput := NewHTTPOutput(replay.URL, &HTTPOutputConfig{Debug: true}) + httpOutput := NewHTTPOutput(replay.URL, &HTTPOutputConfig{}) plugins := &InOutPlugins{ Inputs: []io.Reader{input}, @@ -280,7 +278,7 @@ func TestInputRAWChunkedEncoding(t *testing.T) { } func BenchmarkRAWInputWithReplay(b *testing.B) { - var respCounter, reqCounter, replayCounter, capturedBody uint64 + var respCounter, reqCounter, replayCounter uint64 wg := &sync.WaitGroup{} wg.Add(b.N * 3) // reqCounter + replayCounter + respCounter @@ -307,15 +305,7 @@ func BenchmarkRAWInputWithReplay(b *testing.B) { replay := http.Server{ Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - defer wg.Done() - defer r.Body.Close() - w.Write([]byte("ab")) - atomic.AddUint64(&replayCounter, 1) - data, err := ioutil.ReadAll(r.Body) - if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF { - b.Log(err) - } - atomic.AddUint64(&capturedBody, uint64(len(data))) + wg.Done() }), } go replay.Serve(listener0) @@ -336,10 +326,9 @@ func BenchmarkRAWInputWithReplay(b *testing.B) { } else { atomic.AddUint64(&respCounter, 1) } - atomic.AddUint64(&capturedBody, uint64(len(data))) wg.Done() }) - httpOutput := NewHTTPOutput(replayAddr, &HTTPOutputConfig{Debug: false}) + httpOutput := NewHTTPOutput(replayAddr, &HTTPOutputConfig{}) plugins := &InOutPlugins{ Inputs: []io.Reader{input}, @@ -349,15 +338,9 @@ func BenchmarkRAWInputWithReplay(b *testing.B) { emitter := NewEmitter(quit) go emitter.Start(plugins, Settings.Middleware) now := time.Now() - var buf [1 << 20]byte - buf[1<<20-1] = 'a' - client := NewHTTPClient(originAddr, &HTTPClientConfig{ResponseBufferSize: 2 << 20, CompatibilityMode: true}) + addr := "http://" + originAddr for i := 0; i < b.N; i++ { - if i&1 == 0 { - _, err = client.Get("/") - } else { - _, err = client.Post("/", buf[:]) - } + _, err = http.Get(addr) if err != nil { b.Log(err) wg.Add(-3) @@ -365,6 +348,6 @@ func BenchmarkRAWInputWithReplay(b *testing.B) { } wg.Wait() - b.Logf("%d/%d Requests, %d/%d Responses, %d/%d Replayed, %d Bytes in %s\n", reqCounter, b.N, respCounter, b.N, replayCounter, b.N, capturedBody, time.Since(now)) + b.Logf("%d/%d Requests, %d/%d Responses, %d/%d Replayed in %s\n", reqCounter, b.N, respCounter, b.N, replayCounter, b.N, time.Since(now)) emitter.Close() } diff --git a/kafka.go b/kafka.go index 29c55f1..1c72543 100644 --- a/kafka.go +++ b/kafka.go @@ -12,9 +12,8 @@ import ( "github.com/buger/goreplay/proto" ) -// KafkaConfig should contains required information to +// InputKafkaConfig should contains required information to // build producers. - type InputKafkaConfig struct { producer sarama.AsyncProducer consumer sarama.Consumer @@ -23,6 +22,7 @@ type InputKafkaConfig struct { UseJSON bool `json:"input-kafka-json-format"` } +// OutputKafkaConfig is the representation of kfka output configuration type OutputKafkaConfig struct { producer sarama.AsyncProducer consumer sarama.Consumer @@ -34,8 +34,8 @@ type OutputKafkaConfig struct { // KafkaTLSConfig should contains TLS certificates for connecting to secured Kafka clusters type KafkaTLSConfig struct { CACert string `json:"kafka-tls-ca-cert"` - clientCert string `json:"kafka-tls-client-cert"` - clientKey string `json:"kafka-tls-client-key"` + ClientCert string `json:"kafka-tls-client-cert"` + ClientKey string `json:"kafka-tls-client-key"` } // KafkaMessage should contains catched request information that should be @@ -77,9 +77,9 @@ func NewTLSConfig(clientCertFile, clientKeyFile, caCertFile string) (*tls.Config func NewKafkaConfig(tlsConfig *KafkaTLSConfig) *sarama.Config { config := sarama.NewConfig() // Configuration options go here - if (tlsConfig != nil) && (tlsConfig.CACert != "") && (tlsConfig.clientCert != "") && (tlsConfig.clientKey != "") { + if (tlsConfig != nil) && (tlsConfig.CACert != "") && (tlsConfig.ClientCert != "") && (tlsConfig.ClientKey != "") { config.Net.TLS.Enable = true - tlsConfig, err := NewTLSConfig(tlsConfig.clientCert, tlsConfig.clientKey, tlsConfig.CACert) + tlsConfig, err := NewTLSConfig(tlsConfig.ClientCert, tlsConfig.ClientKey, tlsConfig.CACert) if err != nil { log.Fatal(err) } diff --git a/output_http.go b/output_http.go index 7b9ffc0..7540fed 100644 --- a/output_http.go +++ b/output_http.go @@ -1,56 +1,27 @@ package main import ( + "bufio" + "bytes" + "crypto/tls" "fmt" "io" "log" + "math" + "net/http" + "net/http/httputil" + "net/url" "sync/atomic" "time" - "github.com/buger/goreplay/proto" "github.com/buger/goreplay/size" ) -const initialDynamicWorkers = 10 - -type httpWorker struct { - output *HTTPOutput - client *HTTPClient - lastActivity time.Time - queue chan []byte - stop chan bool -} - -func newHTTPWorker(output *HTTPOutput, queue chan []byte) *httpWorker { - client := NewHTTPClient(output.address, &HTTPClientConfig{ - FollowRedirects: output.config.RedirectLimit, - Debug: output.config.Debug, - OriginalHost: output.config.OriginalHost, - Timeout: output.config.Timeout, - ResponseBufferSize: int(output.config.BufferSize), - }) - - w := &httpWorker{client: client} - if queue == nil { - w.queue = make(chan []byte, 100) - } else { - w.queue = queue - } - w.stop = make(chan bool) - - go func() { - for { - select { - case payload := <-w.queue: - output.sendRequest(client, payload) - case <-w.stop: - return - } - } - }() - - return w -} +const ( + initialDynamicWorkers = 10 + readChunkSize = 64 * 1024 + maxResponseSize = 1073741824 +) type response struct { payload []byte @@ -61,171 +32,136 @@ type response struct { // HTTPOutputConfig struct for holding http output configuration type HTTPOutputConfig struct { - RedirectLimit int `json:"output-http-redirect-limit"` - - Stats bool `json:"output-http-stats"` - WorkersMin int `json:"output-http-workers-min"` - WorkersMax int `json:"output-http-workers"` - StatsMs int `json:"output-http-stats-ms"` - Workers int - QueueLen int `json:"output-http-queue-len"` - - ElasticSearch string `json:"output-http-elasticsearch"` - - Timeout time.Duration `json:"output-http-timeout"` - OriginalHost bool `json:"output-http-original-host"` - BufferSize size.Size `json:"output-http-response-buffer"` - - CompatibilityMode bool `json:"output-http-compatibility-mode"` - - RequestGroup string - - Debug bool `json:"output-http-debug"` - - TrackResponses bool `json:"output-http-track-response"` + TrackResponses bool `json:"output-http-track-response"` + Stats bool `json:"output-http-stats"` + OriginalHost bool `json:"output-http-original-host"` + RedirectLimit int `json:"output-http-redirect-limit"` + WorkersMin int `json:"output-http-workers-min"` + WorkersMax int `json:"output-http-workers"` + StatsMs int `json:"output-http-stats-ms"` + QueueLen int `json:"output-http-queue-len"` + ElasticSearch string `json:"output-http-elasticsearch"` + Timeout time.Duration `json:"output-http-timeout"` + WorkerTimeout time.Duration `json:"output-http-worker-timeout"` + BufferSize size.Size `json:"output-http-response-buffer"` + SkipVerify bool `json:"output-http-skip-verify"` + rawURL string + url *url.URL } // HTTPOutput plugin manage pool of workers which send request to replayed server -// By default workers pool is dynamic and starts with 10 workers -// You can specify fixed number of workers using `--output-http-workers` +// By default workers pool is dynamic and starts with 1 worker or workerMin workers +// You can specify maximum number of workers using `--output-http-workers` type HTTPOutput struct { - // Keep this as first element of struct because it guarantees 64bit - // alignment. atomic.* functions crash on 32bit machines if operand is not - // aligned at 64bit. See https://github.com/golang/go/issues/599 - activeWorkers int64 - - workerSessions map[string]*httpWorker - - address string - limit int - queue chan []byte - - responses chan response - - needWorker chan int - - config *HTTPOutputConfig - - queueStats *GorStat - + activeWorkers int32 + config *HTTPOutputConfig + queueStats *GorStat elasticSearch *ESPlugin - - stop chan bool // Channel used only to indicate goroutine should shutdown + client *HTTPClient + stopWorker chan struct{} + queue chan []byte + responses chan response + stop chan bool // Channel used only to indicate goroutine should shutdown } // NewHTTPOutput constructor for HTTPOutput // Initialize workers func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer { o := new(HTTPOutput) - - o.address = address + var err error + config.url, err = url.Parse(address) + if err != nil { + log.Fatal(fmt.Sprintf("[OUTPUT-HTTP] parse HTTP output URL error[%q]", err)) + } + if config.url.Scheme == "" { + config.url.Scheme = "http" + } + config.rawURL = config.url.String() + if config.Timeout < time.Millisecond*100 { + config.Timeout = time.Second + } + if config.BufferSize <= 0 { + config.BufferSize = 100 * 1024 // 100kb + } + if config.WorkersMin <= 0 { + config.WorkersMin = 1 + } + if config.WorkersMin > 1000 { + config.WorkersMin = 1000 + } + if config.WorkersMax <= 0 { + config.WorkersMax = math.MaxInt32 // idealy so large + } + if config.WorkersMax < config.WorkersMin { + config.WorkersMax = config.WorkersMin + } + if config.QueueLen <= 0 { + config.QueueLen = 1000 + } + if config.RedirectLimit < 0 { + config.RedirectLimit = 0 + } + if config.WorkerTimeout <= 0 { + config.WorkerTimeout = time.Second * 2 + } o.config = config o.stop = make(chan bool) - if o.config.Stats { o.queueStats = NewGorStat("output_http", o.config.StatsMs) } o.queue = make(chan []byte, o.config.QueueLen) o.responses = make(chan response, o.config.QueueLen) - o.needWorker = make(chan int, 1) - - // Initial workers count - if o.config.WorkersMax == 0 { - o.needWorker <- initialDynamicWorkers - } else { - o.needWorker <- o.config.WorkersMax - } + // it should not be buffered to avoid races + o.stopWorker = make(chan struct{}) if o.config.ElasticSearch != "" { o.elasticSearch = new(ESPlugin) o.elasticSearch.Init(o.config.ElasticSearch) } - - if Settings.RecognizeTCPSessions { - if !PRO { - log.Fatal("Detailed TCP sessions work only with PRO license") - } - o.workerSessions = make(map[string]*httpWorker, 100) - go o.sessionWorkerMaster() - } else { - go o.workerMaster() + o.client = NewHTTPClient(o.config) + o.activeWorkers += int32(o.config.WorkersMin) + for i := 0; i < o.config.WorkersMin; i++ { + go o.startWorker() } - + go o.workerMaster() return o } func (o *HTTPOutput) workerMaster() { - for { - newWorkers := <-o.needWorker - atomic.AddInt64(&o.activeWorkers, int64(newWorkers)) - for i := 0; i < newWorkers; i++ { - go o.startWorker() - } - } -} - -func (o *HTTPOutput) sessionWorkerMaster() { - gc := time.Tick(time.Second) - - for { - select { - case p := <-o.queue: - id := payloadID(p) - sessionID := string(id[0:20]) - worker, ok := o.workerSessions[sessionID] - - if !ok { - atomic.AddInt64(&o.activeWorkers, 1) - worker = newHTTPWorker(o, nil) - o.workerSessions[sessionID] = worker - } - - worker.queue <- p - worker.lastActivity = time.Now() - case <-gc: - now := time.Now() - - for id, w := range o.workerSessions { - if !w.lastActivity.IsZero() && now.Sub(w.lastActivity) >= 120*time.Second { - w.stop <- true - delete(o.workerSessions, id) - atomic.AddInt64(&o.activeWorkers, -1) - } - } - } - } -} - -func (o *HTTPOutput) startWorker() { - client := NewHTTPClient(o.address, &HTTPClientConfig{ - FollowRedirects: o.config.RedirectLimit, - Debug: o.config.Debug, - OriginalHost: o.config.OriginalHost, - Timeout: o.config.Timeout, - ResponseBufferSize: int(o.config.BufferSize), - CompatibilityMode: o.config.CompatibilityMode, - }) - + var timer = time.NewTimer(o.config.WorkerTimeout) + defer func() { + // recover from panics caused by trying to send in + // a closed chan(o.stopWorker) + recover() + }() + defer timer.Stop() for { select { case <-o.stop: return + default: + <-timer.C + } + // rollback workers + rollback: + if atomic.LoadInt32(&o.activeWorkers) > int32(o.config.WorkersMin) && len(o.queue) < 1 { + // close one worker + o.stopWorker <- struct{}{} + atomic.AddInt32(&o.activeWorkers, -1) + goto rollback + } + timer.Reset(o.config.WorkerTimeout) + } +} + +func (o *HTTPOutput) startWorker() { + for { + select { + case <-o.stopWorker: + return case data := <-o.queue: - o.sendRequest(client, data) - case <-time.After(2 * time.Second): - // When dynamic scaling enabled workers die after 2s of inactivity - if o.config.WorkersMin == o.config.WorkersMax { - continue - } - - workersCount := int(atomic.LoadInt64(&o.activeWorkers)) - - // At least 1 startWorker should be alive - if workersCount != 1 && workersCount > o.config.WorkersMin { - atomic.AddInt64(&o.activeWorkers, -1) - return - } + o.sendRequest(o.client, data) } } } @@ -237,7 +173,6 @@ func (o *HTTPOutput) Write(data []byte) (n int, err error) { buf := make([]byte, len(data)) copy(buf, data) - select { case <-o.stop: return 0, ErrorStopped @@ -247,22 +182,13 @@ func (o *HTTPOutput) Write(data []byte) (n int, err error) { if o.config.Stats { o.queueStats.Write(len(o.queue)) } - - if !Settings.RecognizeTCPSessions && o.config.WorkersMax != o.config.WorkersMin { - workersCount := int(atomic.LoadInt64(&o.activeWorkers)) - - if len(o.queue) > workersCount { - extraWorkersReq := len(o.queue) - workersCount + 1 - maxWorkersAvailable := o.config.WorkersMax - workersCount - if extraWorkersReq > maxWorkersAvailable { - extraWorkersReq = maxWorkersAvailable - } - if extraWorkersReq > 0 { - o.needWorker <- extraWorkersReq - } + if len(o.queue) > 0 { + // try to start a new worker to serve + if atomic.LoadInt32(&o.activeWorkers) < int32(o.config.WorkersMax) { + go o.startWorker() + atomic.AddInt32(&o.activeWorkers, 1) } } - return len(data), nil } @@ -274,8 +200,6 @@ func (o *HTTPOutput) Read(data []byte) (int, error) { case resp = <-o.responses: } - Debug(3, "[OUTPUT-HTTP] Received response:", string(resp.payload)) - header := payloadHeader(ReplayedResponsePayload, resp.uuid, resp.roundTripTime, resp.startedAt) n := copy(data, header) if len(data) > len(header) { @@ -283,33 +207,25 @@ func (o *HTTPOutput) Read(data []byte) (int, error) { } dis := len(header) + len(data) - n if dis > 0 { - Debug(2, "[OUTPUT-HTTP] discarded", dis, "increase copy buffer size") + Debug(2, fmt.Sprintf("[OUTPUT-HTTP] %dB discarded increase copy buffer size", dis)) } return n, nil } func (o *HTTPOutput) sendRequest(client *HTTPClient, request []byte) { - meta := payloadMeta(request) - - Debug(2, fmt.Sprintf("[OUTPUT-HTTP] meta: %q", meta)) - - if len(meta) < 2 { + if !isRequestPayload(request) { return } - uuid := meta[1] - + uuid := payloadID(request) body := payloadBody(request) - if !proto.HasRequestTitle(body) { - return - } - start := time.Now() resp, err := client.Send(body) stop := time.Now() if err != nil { - Debug(1, "Error when sending ", err) + Debug(1, fmt.Sprintf("[HTTP-OUTPUT] error when sending: %q", err)) + return } if o.config.TrackResponses { @@ -322,11 +238,81 @@ func (o *HTTPOutput) sendRequest(client *HTTPClient, request []byte) { } func (o *HTTPOutput) String() string { - return "HTTP output: " + o.address + return "HTTP output: " + o.config.rawURL } // Close closes the data channel so that data func (o *HTTPOutput) Close() error { close(o.stop) + close(o.stopWorker) return nil } + +// HTTPClient holds configurations for a single HTTP client +type HTTPClient struct { + config *HTTPOutputConfig + Client *http.Client +} + +// NewHTTPClient returns new http client with check redirects policy +func NewHTTPClient(config *HTTPOutputConfig) *HTTPClient { + client := new(HTTPClient) + client.config = config + var transport *http.Transport + client.Client = &http.Client{ + Timeout: client.config.Timeout, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + if len(via) >= client.config.RedirectLimit { + Debug(1, fmt.Sprintf("[HTTPCLIENT] maximum output-http-redirects[%d] reached!", client.config.RedirectLimit)) + return http.ErrUseLastResponse + } + lastReq := via[len(via)-1] + resp := req.Response + Debug(2, fmt.Sprintf("[HTTPCLIENT] HTTP redirects from %q to %q with %q", lastReq.Host, req.Host, resp.Status)) + return nil + }, + } + if config.SkipVerify { + // clone to avoid modying global default RoundTripper + transport = http.DefaultTransport.(*http.Transport).Clone() + transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} + client.Client.Transport = transport + } + + return client +} + +// Send sends an http request using client create by NewHTTPClient +func (c *HTTPClient) Send(data []byte) ([]byte, error) { + var req *http.Request + var resp *http.Response + var err error + + req, err = http.ReadRequest(bufio.NewReader(bytes.NewReader(data))) + if err != nil { + return nil, err + } + // we don't send CONNECT or OPTIONS request + if req.Method == http.MethodConnect { + return nil, nil + } + + if !c.config.OriginalHost { + req.Host = c.config.url.Host + } + + req.URL = c.config.url + // force connection to not be closed, which can affect the global client + req.Close = false + // it's an error if this is not equal to empty string + req.RequestURI = "" + + resp, err = c.Client.Do(req) + if err != nil { + return nil, err + } + if c.config.TrackResponses { + return httputil.DumpResponse(resp, true) + } + return nil, nil +} diff --git a/output_http_test.go b/output_http_test.go index 77183f9..279e4ed 100644 --- a/output_http_test.go +++ b/output_http_test.go @@ -8,7 +8,6 @@ import ( _ "net/http/httputil" "sync" "testing" - "time" ) func TestHTTPOutput(t *testing.T) { @@ -43,16 +42,16 @@ func TestHTTPOutput(t *testing.T) { methods := HTTPMethods{[]byte("GET"), []byte("PUT"), []byte("POST")} Settings.ModifierConfig = HTTPModifierConfig{Headers: headers, Methods: methods} - http_output := NewHTTPOutput(server.URL, &HTTPOutputConfig{Debug: true, TrackResponses: true}) + httpOutput := NewHTTPOutput(server.URL, &HTTPOutputConfig{TrackResponses: true}) output := NewTestOutput(func(data []byte) { wg.Done() }) plugins := &InOutPlugins{ Inputs: []io.Reader{input}, - Outputs: []io.Writer{http_output, output}, + Outputs: []io.Writer{httpOutput, output}, } - plugins.All = append(plugins.All, input, output, http_output) + plugins.All = append(plugins.All, input, output, httpOutput) emitter := NewEmitter(quit) go emitter.Start(plugins, Settings.Middleware) @@ -89,7 +88,7 @@ func TestHTTPOutputKeepOriginalHost(t *testing.T) { headers := HTTPHeaders{HTTPHeader{"Host", "custom-host.com"}} Settings.ModifierConfig = HTTPModifierConfig{Headers: headers} - output := NewHTTPOutput(server.URL, &HTTPOutputConfig{Debug: false, OriginalHost: true}) + output := NewHTTPOutput(server.URL, &HTTPOutputConfig{OriginalHost: true, SkipVerify: true}) plugins := &InOutPlugins{ Inputs: []io.Reader{input}, @@ -118,7 +117,7 @@ func TestHTTPOutputSSL(t *testing.T) { })) input := NewTestInput() - output := NewHTTPOutput(server.URL, &HTTPOutputConfig{}) + output := NewHTTPOutput(server.URL, &HTTPOutputConfig{SkipVerify: true}) plugins := &InOutPlugins{ Inputs: []io.Reader{input}, @@ -151,7 +150,7 @@ func TestHTTPOutputSessions(t *testing.T) { defer server.Close() Settings.RecognizeTCPSessions = true - output := NewHTTPOutput(server.URL, &HTTPOutputConfig{Debug: true}) + output := NewHTTPOutput(server.URL, &HTTPOutputConfig{}) plugins := &InOutPlugins{ Inputs: []io.Reader{input}, @@ -186,14 +185,43 @@ func BenchmarkHTTPOutput(b *testing.B) { wg := new(sync.WaitGroup) quit := make(chan int) - server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - time.Sleep(50 * time.Millisecond) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { wg.Done() })) defer server.Close() input := NewTestInput() - output := NewHTTPOutput(server.URL, &HTTPOutputConfig{}) + output := NewHTTPOutput(server.URL, &HTTPOutputConfig{WorkersMax: 1}) + + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } + plugins.All = append(plugins.All, input, output) + + emitter := NewEmitter(quit) + go emitter.Start(plugins, Settings.Middleware) + + for i := 0; i < b.N; i++ { + wg.Add(1) + input.EmitPOST() + } + + wg.Wait() + emitter.Close() +} + +func BenchmarkHTTPOutputTLS(b *testing.B) { + wg := new(sync.WaitGroup) + quit := make(chan int) + + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + wg.Done() + })) + defer server.Close() + + input := NewTestInput() + output := NewHTTPOutput(server.URL, &HTTPOutputConfig{SkipVerify: true, WorkersMax: 1}) plugins := &InOutPlugins{ Inputs: []io.Reader{input}, diff --git a/output_kafka.go b/output_kafka.go index d030ca2..ecf46b6 100644 --- a/output_kafka.go +++ b/output_kafka.go @@ -7,6 +7,7 @@ import ( "strings" "time" + "github.com/buger/goreplay/byteutils" "github.com/buger/goreplay/proto" "github.com/Shopify/sarama" @@ -71,27 +72,28 @@ func (o *KafkaOutput) Write(data []byte) (n int, err error) { var message sarama.StringEncoder if !o.config.UseJSON { - message = sarama.StringEncoder(data) + message = sarama.StringEncoder(byteutils.SliceToString(data)) } else { - headers := make(map[string]string) - proto.ParseHeaders([][]byte{data}, func(header []byte, value []byte) { - headers[string(header)] = string(value) - }) + mimeHeader := proto.ParseHeaders(data) + var header map[string]string + for k, v := range mimeHeader { + header[k] = strings.Join(v, ", ") + } meta := payloadMeta(data) req := payloadBody(data) kafkaMessage := KafkaMessage{ - ReqURL: string(proto.Path(req)), - ReqType: string(meta[0]), - ReqID: string(meta[1]), - ReqTs: string(meta[2]), - ReqMethod: string(proto.Method(req)), - ReqBody: string(proto.Body(req)), - ReqHeaders: headers, + ReqURL: byteutils.SliceToString(proto.Path(req)), + ReqType: byteutils.SliceToString(meta[0]), + ReqID: byteutils.SliceToString(meta[1]), + ReqTs: byteutils.SliceToString(meta[2]), + ReqMethod: byteutils.SliceToString(proto.Method(req)), + ReqBody: byteutils.SliceToString(proto.Body(req)), + ReqHeaders: header, } jsonMessage, _ := json.Marshal(&kafkaMessage) - message = sarama.StringEncoder(jsonMessage) + message = sarama.StringEncoder(byteutils.SliceToString(jsonMessage)) } o.producer.Input() <- &sarama.ProducerMessage{ diff --git a/proto/fuzz.go b/proto/fuzz.go index 0220c7a..e257940 100644 --- a/proto/fuzz.go +++ b/proto/fuzz.go @@ -4,9 +4,7 @@ package proto func Fuzz(data []byte) int { - ParseHeaders([][]byte{data}, func(header []byte, value []byte) bool { - return true - }) + ParseHeaders(data) return 1 } diff --git a/proto/proto.go b/proto/proto.go index 5fd954f..c634ac2 100644 --- a/proto/proto.go +++ b/proto/proto.go @@ -111,14 +111,13 @@ func header(payload []byte, name []byte) (value []byte, headerStart, headerEnd, return } -// ParseHeaders Parsing headers from multiple payloads -func ParseHeaders(payloads [][]byte, cb func(header []byte, value []byte)) { - p := bytes.Join(payloads, nil) +// ParseHeaders Parsing headers from the payload +func ParseHeaders(p []byte) textproto.MIMEHeader { // trimming off the title of the request - if HasRequestTitle(p) || HasResponseTitle(p) { + if HasTitle(p) { headerStart := MIMEHeadersStartPos(p) if headerStart > len(p)-1 { - return + return nil } p = p[headerStart:] } @@ -126,17 +125,12 @@ func ParseHeaders(payloads [][]byte, cb func(header []byte, value []byte)) { if headerEnd > 1 { p = p[:headerEnd] } - reader := textproto.NewReader(bufio.NewReader(bytes.NewBuffer(p))) + reader := textproto.NewReader(bufio.NewReader(bytes.NewReader(p))) mime, err := reader.ReadMIMEHeader() if err != nil { - return + return nil } - for k, v := range mime { - for _, value := range v { - cb([]byte(k), []byte(value)) - } - } - return + return mime } // Header returns header value, if header not found, value will be blank diff --git a/proto/proto_test.go b/proto/proto_test.go index a5675a1..0e012bc 100644 --- a/proto/proto_test.go +++ b/proto/proto_test.go @@ -2,6 +2,7 @@ package proto import ( "bytes" + "net/textproto" "reflect" "testing" ) @@ -124,16 +125,12 @@ func TestDeleteHeader(t *testing.T) { func TestParseHeaders(t *testing.T) { payload := [][]byte{[]byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.or"), []byte("g\r\nUser-Ag"), []byte("ent:Chrome\r\n\r\n"), []byte("Fake-Header: asda")} - headers := make(map[string]string) + headers := ParseHeaders(bytes.Join(payload, nil)) - ParseHeaders(payload, func(header []byte, value []byte) { - headers[string(header)] = string(value) - }) - - expected := map[string]string{ - "Content-Length": "7", - "Host": "www.w3.org", - "User-Agent": "Chrome", + expected := textproto.MIMEHeader{ + "Content-Length": []string{"7"}, + "Host": []string{"www.w3.org"}, + "User-Agent": []string{"Chrome"}, } if !reflect.DeepEqual(headers, expected) { @@ -148,8 +145,7 @@ func TestFuzzCrashers(t *testing.T) { } for _, f := range crashers { - ParseHeaders([][]byte{[]byte(f)}, func(header []byte, value []byte) { - }) + ParseHeaders([]byte(f)) } } @@ -158,17 +154,13 @@ func TestParseHeadersWithComplexUserAgent(t *testing.T) { // Parser should wait for \r\n payload := [][]byte{[]byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.or"), []byte("g\r\nUser-Ag"), []byte("ent:Mozilla/5.0 (Windows NT 6.1; WOW64; Trident/7.0; rv:11.0) like Gecko\r\n\r\n"), []byte("Fake-Header: asda")} - headers := make(map[string]string) - - ParseHeaders(payload, func(header []byte, value []byte) { - headers[string(header)] = string(value) - }) + headers := ParseHeaders(bytes.Join(payload, nil)) expected := map[string]string{ "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64; Trident/7.0; rv:11.0) like Gecko", } - if expected["User-Agent"] != headers["User-Agent"] { + if expected["User-Agent"] != headers["User-Agent"][0] { t.Errorf("Header 'User-Agent' expected '%s' and parsed: '%s'", expected["User-Agent"], headers["User-Agent"]) } } @@ -178,11 +170,7 @@ func TestParseHeadersWithOrigin(t *testing.T) { // Parser should wait for \r\n payload := [][]byte{[]byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.or"), []byte("g\r\nReferrer: http://127.0.0.1:3000\r\nOrigi"), []byte("n: https://www.example.com\r\nUser-Ag"), []byte("ent:Mozilla/5.0 (Windows NT 6.1; WOW64; Trident/7.0; rv:11.0) like Gecko\r\n\r\n"), []byte("in:https://www.example.com\r\n\r\n"), []byte("Fake-Header: asda")} - headers := make(map[string]string) - - ParseHeaders(payload, func(header []byte, value []byte) { - headers[string(header)] = string(value) - }) + headers := ParseHeaders(bytes.Join(payload, nil)) expected := map[string]string{ "Origin": "https://www.example.com", @@ -190,15 +178,15 @@ func TestParseHeadersWithOrigin(t *testing.T) { "Referrer": "http://127.0.0.1:3000", } - if expected["Referrer"] != headers["Referrer"] { + if expected["Referrer"] != headers["Referrer"][0] { t.Errorf("Header 'Referrer' expected '%s' and parsed: '%s'", expected["Referrer"], headers["Referrer"]) } - if expected["Origin"] != headers["Origin"] { + if expected["Origin"] != headers["Origin"][0] { t.Errorf("Header 'Origin' expected '%s' and parsed: '%s'", expected["Origin"], headers["Origin"]) } - if expected["User-Agent"] != headers["User-Agent"] { + if expected["User-Agent"] != headers["User-Agent"][0] { t.Errorf("Header 'User-Agent' expected '%s' and parsed: '%s'", expected["User-Agent"], headers["User-Agent"]) } } diff --git a/protocol.go b/protocol.go index 9f40b6f..5a73152 100644 --- a/protocol.go +++ b/protocol.go @@ -72,19 +72,11 @@ func payloadID(payload []byte) (id []byte) { if len(meta) < 2 { return } - // id is encoded in hex, we need to revert to how it was - id = make([]byte, 20) - hex.Decode(id, meta[1]) - return + return meta[1] } func isOriginPayload(payload []byte) bool { - switch payload[0] { - case RequestPayload, ResponsePayload: - return true - default: - return false - } + return payload[0] == RequestPayload || payload[0] == ResponsePayload } func isRequestPayload(payload []byte) bool { diff --git a/settings.go b/settings.go index 4ebf613..2a52e03 100644 --- a/settings.go +++ b/settings.go @@ -152,11 +152,11 @@ func init() { /* outputHTTPConfig */ flag.Var(&Settings.OutputHTTPConfig.BufferSize, "output-http-response-buffer", "HTTP response buffer size, all data after this size will be discarded.") - flag.BoolVar(&Settings.OutputHTTPConfig.CompatibilityMode, "output-http-compatibility-mode", false, "Use standard Go client, instead of built-in implementation. Can be slower, but more compatible.") - flag.IntVar(&Settings.OutputHTTPConfig.WorkersMin, "output-http-workers-min", 0, "Gor uses dynamic worker scaling. Enter a number to set a minimum number of workers. default = 1.") flag.IntVar(&Settings.OutputHTTPConfig.WorkersMax, "output-http-workers", 0, "Gor uses dynamic worker scaling. Enter a number to set a maximum number of workers. default = 0 = unlimited.") flag.IntVar(&Settings.OutputHTTPConfig.QueueLen, "output-http-queue-len", 1000, "Number of requests that can be queued for output, if all workers are busy. default = 1000") + flag.BoolVar(&Settings.OutputHTTPConfig.SkipVerify, "output-http-skip-verify", false, "Don't verify hostname on TLS secure connection.") + flag.DurationVar(&Settings.OutputHTTPConfig.WorkerTimeout, "output-http-worker-timeout", 2*time.Second, "Duration to rollback idle workers.") flag.IntVar(&Settings.OutputHTTPConfig.RedirectLimit, "output-http-redirects", 0, "Enable how often redirects should be followed.") flag.DurationVar(&Settings.OutputHTTPConfig.Timeout, "output-http-timeout", 5*time.Second, "Specify HTTP request/response timeout. By default 5s. Example: --output-http-timeout 30s") @@ -165,11 +165,11 @@ func init() { flag.BoolVar(&Settings.OutputHTTPConfig.Stats, "output-http-stats", false, "Report http output queue stats to console every N milliseconds. See output-http-stats-ms") flag.IntVar(&Settings.OutputHTTPConfig.StatsMs, "output-http-stats-ms", 5000, "Report http output queue stats to console every N milliseconds. default: 5000") flag.BoolVar(&Settings.OutputHTTPConfig.OriginalHost, "http-original-host", false, "Normally gor replaces the Host http header with the host supplied with --output-http. This option disables that behavior, preserving the original Host header.") - flag.BoolVar(&Settings.OutputHTTPConfig.Debug, "output-http-debug", false, "Enables http debug output.") flag.StringVar(&Settings.OutputHTTPConfig.ElasticSearch, "output-http-elasticsearch", "", "Send request and response stats to ElasticSearch:\n\tgor --input-raw :8080 --output-http staging.com --output-http-elasticsearch 'es_host:api_port/index_name'") /* outputHTTPConfig */ flag.Var(&Settings.OutputBinary, "output-binary", "Forwards incoming binary payloads to given address.\n\t# Redirect all incoming requests to staging.com address \n\tgor --input-raw :80 --input-raw-protocol binary --output-binary staging.com:80") + /* outputBinaryConfig */ flag.Var(&Settings.OutputBinaryConfig.BufferSize, "output-tcp-response-buffer", "TCP response buffer size, all data after this size will be discarded.") flag.IntVar(&Settings.OutputBinaryConfig.Workers, "output-binary-workers", 0, "Gor uses dynamic worker scaling by default. Enter a number to run a set number of workers.") @@ -188,8 +188,8 @@ func init() { flag.BoolVar(&Settings.InputKafkaConfig.UseJSON, "input-kafka-json-format", false, "If turned on, it will assume that messages coming in JSON format rather than GoReplay text format.") flag.StringVar(&Settings.KafkaTLSConfig.CACert, "kafka-tls-ca-cert", "", "CA certificate for Kafka TLS Config:\n\tgor --input-raw :3000 --output-kafka-host '192.168.0.1:9092' --output-kafka-topic 'topic' --kafka-tls-ca-cert cacert.cer.pem --kafka-tls-client-cert client.cer.pem --kafka-tls-client-key client.key.pem") - flag.StringVar(&Settings.KafkaTLSConfig.clientCert, "kafka-tls-client-cert", "", "Client certificate for Kafka TLS Config (mandatory with to kafka-tls-ca-cert and kafka-tls-client-key)") - flag.StringVar(&Settings.KafkaTLSConfig.clientKey, "kafka-tls-client-key", "", "Client Key for Kafka TLS Config (mandatory with to kafka-tls-client-cert and kafka-tls-client-key)") + flag.StringVar(&Settings.KafkaTLSConfig.ClientCert, "kafka-tls-client-cert", "", "Client certificate for Kafka TLS Config (mandatory with to kafka-tls-ca-cert and kafka-tls-client-key)") + flag.StringVar(&Settings.KafkaTLSConfig.ClientKey, "kafka-tls-client-key", "", "Client Key for Kafka TLS Config (mandatory with to kafka-tls-client-cert and kafka-tls-client-key)") flag.Var(&Settings.ModifierConfig.Headers, "http-set-header", "Inject additional headers to http reqest:\n\tgor --input-raw :8080 --output-http staging.com --http-set-header 'User-Agent: Gor'") flag.Var(&Settings.ModifierConfig.Headers, "output-http-header", "WARNING: `--output-http-header` DEPRECATED, use `--http-set-header` instead")