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

deprecate output-http non-compatible clients (#833)

Browse Source 
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!
This commit is contained in:
Urban Ishimwe
2020-10-13 08:36:16 +03:00
committed by GitHub
parent 3635d66a76
commit 5d8ca525a4
14 changed files with 308 additions and 1403 deletions
Download Patch File Download Diff File Expand all files Collapse all files
http_client.go
-530
View File
@@ -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
}
http_client_test.go
-532
View File
@@ -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")
// }
// }
http_prettifier.go
+17 -20
View File
@@ -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"))
input_kafka.go
+2 -3
View File
@@ -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
}
input_raw_test.go
+13 -30
View File
@@ -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()
}
kafka.go
+6 -6
View File
@@ -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)
}
output_http.go
+189 -203
View File
@@ -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
}
output_http_test.go
+38 -10
View File
@@ -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},
output_kafka.go
+15 -13
View File
@@ -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{
proto/fuzz.go
+1 -3
View File
@@ -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
}
proto/proto.go
+7 -13
View File
@@ -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
proto/proto_test.go
+13 -25
View File
@@ -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"])
}
}
protocol.go
+2 -10
View File
@@ -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 {
settings.go
+5 -5
View File
@@ -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")