From 6647481fe07061bc90be6db55611dd34179d7c6c Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 13 Jul 2015 09:31:11 +0500 Subject: [PATCH 1/6] Improve style and docs for proto package --- Makefile | 2 +- elasticsearch.go | 30 +++---- gor.go | 2 +- http_client.go | 4 +- http_modifier.go | 8 +- input_raw_test.go | 2 +- proto/proto.go | 134 +++++++++++++++++++----------- proto/proto_test.go | 46 +++++----- raw_socket_listener/listener.go | 40 ++++----- raw_socket_listener/tcp_packet.go | 2 +- 10 files changed, 154 insertions(+), 116 deletions(-) diff --git a/Makefile b/Makefile index 2f8b95b..cb4bb17 100644 --- a/Makefile +++ b/Makefile @@ -14,7 +14,7 @@ dbuild: docker build -t gor . dlint: - docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor golint $(ARGS) + docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor golint $(PKG) drace: docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor go test ./... $(ARGS) -v -race -timeout 15s diff --git a/elasticsearch.go b/elasticsearch.go index 801e75c..a930ac3 100644 --- a/elasticsearch.go +++ b/elasticsearch.go @@ -122,24 +122,24 @@ func (p *ESPlugin) ResponseAnalyze(req, resp []byte, start, stop time.Time) { esResp := ESRequestResponse{ ReqUrl: proto.Path(req), ReqMethod: proto.Method(req), - ReqUserAgent: proto.GetHeader(req, "User-Agent"), - ReqAcceptLanguage: proto.GetHeader(req, "Accept-Language"), - ReqAccept: proto.GetHeader(req, "Accept"), - ReqAcceptEncoding: proto.GetHeader(req, "Accept-Encoding"), - ReqIfModifiedSince: proto.GetHeader(req, "If-Modified-Since"), - ReqConnection: proto.GetHeader(req, "Connection"), - ReqCookies: proto.GetHeader(req, "Cookie"), + ReqUserAgent: proto.Header(req, []byte("User-Agent")), + ReqAcceptLanguage: proto.Header(req, []byte("Accept-Language")), + ReqAccept: proto.Header(req, []byte("Accept")), + ReqAcceptEncoding: proto.Header(req, []byte("Accept-Encoding")), + ReqIfModifiedSince: proto.Header(req, []byte("If-Modified-Since")), + ReqConnection: proto.Header(req, []byte("Connection")), + ReqCookies: proto.Header(req, []byte("Cookie")), RespStatus: proto.Status(resp), RespStatusCode: proto.Status(resp), RespProto: proto.Method(resp), - RespContentLength: proto.GetHeader(resp, "Content-Length"), - RespContentType: proto.GetHeader(resp, "Content-Type"), - RespTransferEncoding: proto.GetHeader(resp, "Transfer-Encoding"), - RespContentEncoding: proto.GetHeader(resp, "Content-Encoding"), - RespExpires: proto.GetHeader(resp, "Expires"), - RespCacheControl: proto.GetHeader(resp, "Cache-Control"), - RespVary: proto.GetHeader(resp, "Vary"), - RespSetCookie: proto.GetHeader(resp, "Set-Cookie"), + RespContentLength: proto.Header(resp, []byte("Content-Length")), + RespContentType: proto.Header(resp, []byte("Content-Type")), + RespTransferEncoding: proto.Header(resp, []byte("Transfer-Encoding")), + RespContentEncoding: proto.Header(resp, []byte("Content-Encoding")), + RespExpires: proto.Header(resp, []byte("Expires")), + RespCacheControl: proto.Header(resp, []byte("Cache-Control")), + RespVary: proto.Header(resp, []byte("Vary")), + RespSetCookie: proto.Header(resp, []byte("Set-Cookie")), Rtt: rtt, Timestamp: t, } diff --git a/gor.go b/gor.go index c3c30bb..af9bf5f 100644 --- a/gor.go +++ b/gor.go @@ -7,9 +7,9 @@ import ( "fmt" "log" "os" + "runtime" "runtime/debug" "runtime/pprof" - "runtime" "time" ) diff --git a/http_client.go b/http_client.go index dd6a739..c54e6a8 100644 --- a/http_client.go +++ b/http_client.go @@ -147,7 +147,7 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { if status[0] == '3' { c.redirectsCount += 1 - location, _, _, _ := proto.Header(payload, []byte("Location")) + location := proto.Header(payload, []byte("Location")) redirectPayload := []byte("GET " + string(location) + " HTTP/1.1\r\n\r\n") if c.config.Debug { @@ -167,4 +167,4 @@ 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)) -} \ No newline at end of file +} diff --git a/http_modifier.go b/http_modifier.go index d820df6..96be68b 100644 --- a/http_modifier.go +++ b/http_modifier.go @@ -86,9 +86,9 @@ func (m *HTTPModifier) Rewrite(payload []byte) (response []byte) { if len(m.config.headerFilters) > 0 { for _, f := range m.config.headerFilters { - value, s, _, _ := proto.Header(payload, f.name) + value := proto.Header(payload, f.name) - if s != -1 && !f.regexp.Match(value) { + if len(value) > 0 && !f.regexp.Match(value) { return } } @@ -96,9 +96,9 @@ func (m *HTTPModifier) Rewrite(payload []byte) (response []byte) { if len(m.config.headerHashFilters) > 0 { for _, f := range m.config.headerHashFilters { - value, s, _, _ := proto.Header(payload, f.name) + value := proto.Header(payload, f.name) - if s != -1 { + if len(value) > 0 { hasher := fnv.New32a() hasher.Write(value) diff --git a/input_raw_test.go b/input_raw_test.go index 2f8652b..6262667 100644 --- a/input_raw_test.go +++ b/input_raw_test.go @@ -31,7 +31,7 @@ func TestRAWInput(t *testing.T) { address := strings.Replace(listener.Addr().String(), "[::]", "127.0.0.1", -1) - client := NewHTTPClient(address, &HTTPClientConfig{}) + client := NewHTTPClient(address, &HTTPClientConfig{}) go Start(quit) diff --git a/proto/proto.go b/proto/proto.go index ee4a1a0..f595234 100644 --- a/proto/proto.go +++ b/proto/proto.go @@ -1,28 +1,50 @@ -// Low-level interaction with HTTP request payload +/* +Package proto provides byte-level interaction with HTTP request payload. + +Example of HTTP payload for future references, new line symbols escaped: + + POST /upload HTTP/1.1\r\n + User-Agent: Gor\r\n + Content-Length: 11\r\n + \r\n + Hello world + + GET /index.html HTTP/1.1\r\n + User-Agent: Gor\r\n + \r\n + \r\n +*/ package proto import ( "bytes" "github.com/buger/gor/byteutils" - _ "log" ) +// In HTTP newline defined by 2 bytes (for both windows and *nix support) var CLRF = []byte("\r\n") -var EMPTY_LINE = []byte("\r\n\r\n") -var HEADER_DELIM = []byte(": ") -// Headers should end with empty line +// New line acts as separator: end of Headers or Body (in some cases) +var EmptyLine = []byte("\r\n\r\n") + +// Separator for Header line. Header looks like: `HeaderName: value` +var HeaderDelim = []byte(": ") + +// MIMEHeadersEndPos finds end of the Headers section, which should end with empty line. func MIMEHeadersEndPos(payload []byte) int { - return bytes.Index(payload, EMPTY_LINE) + return bytes.Index(payload, EmptyLine) } +// MIMEHeadersStartPos finds start of Headers section +// It just finds position of second line (first contains location and method). func MIMEHeadersStartPos(payload []byte) int { return bytes.Index(payload, CLRF) + 2 // Find first line end } -// Find header value or return error -// Do not support multi-line headers -func Header(payload []byte, name []byte) (value []byte, headerStart, valueStart, headerEnd int) { +// header return value and positions of header/value start/end. +// If not found, value will be blank, and headerStart will be -1 +// Do not support multi-line headers. +func header(payload []byte, name []byte) (value []byte, headerStart, valueStart, headerEnd int) { headerStart = bytes.Index(payload, name) if headerStart == -1 { @@ -31,7 +53,7 @@ func Header(payload []byte, name []byte) (value []byte, headerStart, valueStart, valueStart = headerStart + len(name) + 1 // Skip ":" after header name if payload[valueStart] == ' ' { // Ignore empty space after ':' - valueStart += 1 + valueStart++ } headerEnd = valueStart + bytes.IndexByte(payload[valueStart:], '\r') value = payload[valueStart:headerEnd] @@ -39,27 +61,32 @@ func Header(payload []byte, name []byte) (value []byte, headerStart, valueStart, return } -func GetHeader(payload []byte, name string) []byte { - val, _, _, _ := Header(payload, []byte(name)) +// Header returns header value, if header not found, value will be blank +func Header(payload, name []byte) []byte { + val, _, _, _ := header(payload, name) return val } +// SetHeader sets header value. If header not found it creates new one. +// Returns modified request payload func SetHeader(payload, name, value []byte) []byte { - _, hs, vs, he := Header(payload, name) + _, hs, vs, he := header(payload, name) - // If header found if hs != -1 { + // If header found we just repace its value return byteutils.Replace(payload, vs, he, value) - } else { - return AddHeader(payload, name, value) } + + return AddHeader(payload, name, value) } +// AddHeader takes http payload and appends new header to the start of headers section +// Returns modified request payload func AddHeader(payload, name, value []byte) []byte { header := make([]byte, len(name)+2+len(value)+2) copy(header[0:], name) - copy(header[len(name):], HEADER_DELIM) + copy(header[len(name):], HeaderDelim) copy(header[len(name)+2:], value) copy(header[len(header)-2:], CLRF) @@ -68,30 +95,31 @@ func AddHeader(payload, name, value []byte) []byte { return byteutils.Insert(payload, mimeStart, header) } +// Path takes payload and retuns request path: Split(firstLine, ' ')[1] func Path(payload []byte) []byte { - start := bytes.IndexByte(payload, ' ') - start += 1 - + start := bytes.IndexByte(payload, ' ') + 1 end := bytes.IndexByte(payload[start:], ' ') return payload[start : start+end] } +// SetPath takes payload, sets new path and returns modified payload func SetPath(payload, path []byte) []byte { - start := bytes.IndexByte(payload, ' ') - start += 1 - + start := bytes.IndexByte(payload, ' ') + 1 end := bytes.IndexByte(payload[start:], ' ') return byteutils.Replace(payload, start, start+end, path) } +// PathParam returns URL query attribute by given name, if no found: valueStart will be -1 func PathParam(payload, name []byte) (value []byte, valueStart, valueEnd int) { path := Path(payload) if paramStart := bytes.Index(path, append(name, '=')); paramStart != -1 { valueStart := paramStart + len(name) + 1 paramEnd := bytes.IndexByte(path[valueStart:], '&') + + // Param can end with '&' (another param), or end of line if paramEnd == -1 { // It is final param paramEnd = len(path) } else { @@ -99,43 +127,51 @@ func PathParam(payload, name []byte) (value []byte, valueStart, valueEnd int) { } return path[valueStart:paramEnd], valueStart, paramEnd - } else { - return []byte(""), -1, -1 } + + return []byte(""), -1, -1 } +// SetPathParam takes payload and updates path Query attribute +// If query param not found, it will append new +// Returns modified payload func SetPathParam(payload, name, value []byte) []byte { path := Path(payload) _, vs, ve := PathParam(payload, name) - if vs != -1 { + if vs != -1 { // If param found, replace its value and set new Path newPath := make([]byte, len(path)) copy(newPath, path) newPath = byteutils.Replace(newPath, vs, ve, value) - return SetPath(payload, newPath) - } else { // if param not found append to end of url - // Adding 2 because of '?' or '&' at start, and '=' in middle - newParam := make([]byte, len(name)+len(value)+2) - - if bytes.IndexByte(path, '?') == -1 { - newParam[0] = '?' - } else { - newParam[0] = '&' - } - - copy(newParam[1:], name) - newParam[1+len(name)] = '=' - copy(newParam[2+len(name):], value) - - newPath := make([]byte, len(path)+len(newParam)) - copy(newPath, path) - copy(newPath[len(path):], newParam) - return SetPath(payload, newPath) } + + // if param not found append to end of url + // Adding 2 because of '?' or '&' at start, and '=' in middle + newParam := make([]byte, len(name)+len(value)+2) + + if bytes.IndexByte(path, '?') == -1 { + newParam[0] = '?' + } else { + newParam[0] = '&' + } + + // Copy "param=value" into buffer, after it looks like "?param=value" + copy(newParam[1:], name) + newParam[1+len(name)] = '=' + copy(newParam[2+len(name):], value) + + // Append param to the end of path + newPath := make([]byte, len(path)+len(newParam)) + copy(newPath, path) + copy(newPath[len(path):], newParam) + + return SetPath(payload, newPath) } +// SetHost updates Host header for HTTP/1.1 or updates host in path for HTTP/1.0 or Proxy requests +// Returns modified payload func SetHost(payload, url, host []byte) []byte { // If this is HTTP 1.0 traffic or proxy traffic it may include host right into path variable, so instead of setting Host header we rewrite Path // Fix for https://github.com/buger/gor/issues/156 @@ -149,18 +185,20 @@ func SetHost(payload, url, host []byte) []byte { newPath = byteutils.Replace(newPath, 0, hostEnd, url) return SetPath(payload, newPath) - } else { - return SetHeader(payload, []byte("Host"), host) } + + return SetHeader(payload, []byte("Host"), host) } +// Method returns HTTP method func Method(payload []byte) []byte { end := bytes.IndexByte(payload, ' ') return payload[:end] } -// Status in response have same position as Path in request +// Status returns response status. +// It happend to be in same position as request payload path func Status(payload []byte) []byte { return Path(payload) } diff --git a/proto/proto_test.go b/proto/proto_test.go index f232ee3..80adbe2 100644 --- a/proto/proto_test.go +++ b/proto/proto_test.go @@ -11,17 +11,17 @@ func TestHeader(t *testing.T) { payload = []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - if val, _, _, _ = Header(payload, []byte("Content-Length")); !bytes.Equal(val, []byte("7")) { + if val = Header(payload, []byte("Content-Length")); !bytes.Equal(val, []byte("7")) { t.Error("Should find header value") } payload = []byte("POST /post HTTP/1.1\r\nContent-Length:7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - if val, _, _, _ = Header(payload, []byte("Content-Length")); !bytes.Equal(val, []byte("7")) { + if val = Header(payload, []byte("Content-Length")); !bytes.Equal(val, []byte("7")) { t.Error("Should find header value without space after :") } - if _, headerStart, _, _ = Header(payload, []byte("Not-Found")); headerStart != -1 { + if _, headerStart, _, _ = header(payload, []byte("Not-Found")); headerStart != -1 { t.Error("Should not found header") } } @@ -50,19 +50,19 @@ func TestMIMEHeadersStartPos(t *testing.T) { } func TestSetHeader(t *testing.T) { - var payload, payload_after []byte + var payload, payloadAfter []byte payload = []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - payload_after = []byte("POST /post HTTP/1.1\r\nContent-Length: 14\r\nHost: www.w3.org\r\n\r\na=1&b=2") + payloadAfter = []byte("POST /post HTTP/1.1\r\nContent-Length: 14\r\nHost: www.w3.org\r\n\r\na=1&b=2") - if payload = SetHeader(payload, []byte("Content-Length"), []byte("14")); !bytes.Equal(payload, payload_after) { + if payload = SetHeader(payload, []byte("Content-Length"), []byte("14")); !bytes.Equal(payload, payloadAfter) { t.Error("Should update header if it exists", string(payload)) } payload = []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - payload_after = []byte("POST /post HTTP/1.1\r\nUser-Agent: Gor\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") + payloadAfter = []byte("POST /post HTTP/1.1\r\nUser-Agent: Gor\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - if payload = SetHeader(payload, []byte("User-Agent"), []byte("Gor")); !bytes.Equal(payload, payload_after) { + if payload = SetHeader(payload, []byte("User-Agent"), []byte("Gor")); !bytes.Equal(payload, payloadAfter) { t.Error("Should add header if not found", string(payload)) } } @@ -78,12 +78,12 @@ func TestPath(t *testing.T) { } func TestSetPath(t *testing.T) { - var payload, payload_after []byte + var payload, payloadAfter []byte payload = []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - payload_after = []byte("POST /new_path HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") + payloadAfter = []byte("POST /new_path HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - if payload = SetPath(payload, []byte("/new_path")); !bytes.Equal(payload, payload_after) { + if payload = SetPath(payload, []byte("/new_path")); !bytes.Equal(payload, payloadAfter) { t.Error("Should replace path", string(payload)) } } @@ -103,44 +103,44 @@ func TestPathParam(t *testing.T) { } func TestSetPathParam(t *testing.T) { - var payload, payload_after []byte + var payload, payloadAfter []byte payload = []byte("POST /post?param=test&user_id=1 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - payload_after = []byte("POST /post?param=new&user_id=1 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") + payloadAfter = []byte("POST /post?param=new&user_id=1 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - if payload = SetPathParam(payload, []byte("param"), []byte("new")); !bytes.Equal(payload, payload_after) { + if payload = SetPathParam(payload, []byte("param"), []byte("new")); !bytes.Equal(payload, payloadAfter) { t.Error("Should replace existing value", string(payload)) } payload = []byte("POST /post?param=test&user_id=1 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - payload_after = []byte("POST /post?param=test&user_id=2 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") + payloadAfter = []byte("POST /post?param=test&user_id=2 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - if payload = SetPathParam(payload, []byte("user_id"), []byte("2")); !bytes.Equal(payload, payload_after) { + if payload = SetPathParam(payload, []byte("user_id"), []byte("2")); !bytes.Equal(payload, payloadAfter) { t.Error("Should replace existing value", string(payload)) } payload = []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - payload_after = []byte("POST /post?param=test HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") + payloadAfter = []byte("POST /post?param=test HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - if payload = SetPathParam(payload, []byte("param"), []byte("test")); !bytes.Equal(payload, payload_after) { + if payload = SetPathParam(payload, []byte("param"), []byte("test")); !bytes.Equal(payload, payloadAfter) { t.Error("Should set param if url have no params", string(payload)) } payload = []byte("POST /post?param=test HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - payload_after = []byte("POST /post?param=test&user_id=1 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") + payloadAfter = []byte("POST /post?param=test&user_id=1 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - if payload = SetPathParam(payload, []byte("user_id"), []byte("1")); !bytes.Equal(payload, payload_after) { + if payload = SetPathParam(payload, []byte("user_id"), []byte("1")); !bytes.Equal(payload, payloadAfter) { t.Error("Should set param at the end if url params", string(payload)) } } func TestSetHostHTTP10(t *testing.T) { - var payload, payload_after []byte + var payload, payloadAfter []byte payload = []byte("POST http://example.com/post HTTP/1.0\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - payload_after = []byte("POST http://new.com/post HTTP/1.0\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") + payloadAfter = []byte("POST http://new.com/post HTTP/1.0\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - if payload = SetHost(payload, []byte("http://new.com"), []byte("new.com")); !bytes.Equal(payload, payload_after) { + if payload = SetHost(payload, []byte("http://new.com"), []byte("new.com")); !bytes.Equal(payload, payloadAfter) { t.Error("Should replace host", string(payload)) } } diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index dce1dc0..9fd0c65 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -69,34 +69,34 @@ func (t *Listener) listen() { } } func (t *Listener) readRAWSocket() { - conn, e := net.ListenPacket("ip4:tcp", t.addr) + conn, e := net.ListenPacket("ip4:tcp", t.addr) - if e != nil { - log.Fatal(e) - } + if e != nil { + log.Fatal(e) + } - defer conn.Close() + defer conn.Close() - for { - buf := make([]byte, 64*1024) // 64kb - // Note: ReadFrom receive messages without IP header - n, addr, err := conn.ReadFrom(buf) + for { + buf := make([]byte, 64*1024) // 64kb + // Note: ReadFrom receive messages without IP header + n, addr, err := conn.ReadFrom(buf) - if err != nil { - log.Println("Error:", err) - continue - } + if err != nil { + log.Println("Error:", err) + continue + } - if n > 0 { - go t.parsePacket(addr, buf[:n]) - } - } + if n > 0 { + go t.parsePacket(addr, buf[:n]) + } + } } func (t *Listener) parsePacket(addr net.Addr, buf []byte) { - if t.isIncomingDataPacket(buf) { - t.c_packets <- ParseTCPPacket(addr, buf) - } + if t.isIncomingDataPacket(buf) { + t.c_packets <- ParseTCPPacket(addr, buf) + } } func (t *Listener) isIncomingDataPacket(buf []byte) bool { diff --git a/raw_socket_listener/tcp_packet.go b/raw_socket_listener/tcp_packet.go index b07109b..4d3f4e6 100644 --- a/raw_socket_listener/tcp_packet.go +++ b/raw_socket_listener/tcp_packet.go @@ -59,7 +59,7 @@ func (t *TCPPacket) Parse() { // ParseBasic set of fields func (t *TCPPacket) ParseBasic() { - t.SrcPort = binary.BigEndian.Uint16(t.Data[0:2]) + t.SrcPort = binary.BigEndian.Uint16(t.Data[0:2]) t.Seq = binary.BigEndian.Uint32(t.Data[4:8]) t.Ack = binary.BigEndian.Uint32(t.Data[8:12]) t.DataOffset = (t.Data[12] & 0xF0) >> 4 From 32081b2d1791d41204b783e24993693858f3d7b0 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 13 Jul 2015 09:36:54 +0500 Subject: [PATCH 2/6] Improve byteutils --- byteutils/byteutils.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/byteutils/byteutils.go b/byteutils/byteutils.go index c65bbdb..f1c1e15 100644 --- a/byteutils/byteutils.go +++ b/byteutils/byteutils.go @@ -1,5 +1,7 @@ +// Package byteutils probvides helpers for working with byte slices package byteutils +// Cut elements from slice for a given range func Cut(a []byte, from, to int) []byte { copy(a[from:], a[to:]) a = a[:len(a)-to+from] @@ -7,6 +9,7 @@ func Cut(a []byte, from, to int) []byte { return a } +// Insert new slice at specified position func Insert(a []byte, i int, b []byte) []byte { a = append(a, make([]byte, len(b))...) copy(a[i+len(b):], a[i:]) @@ -15,7 +18,7 @@ func Insert(a []byte, i int, b []byte) []byte { return a } -// Unlike bytes.Replace it allows you to specify range +// Replace function unlike bytes.Replace allows you to specify range func Replace(a []byte, from, to int, new []byte) []byte { lenDiff := len(new) - (to - from) From 0859642b386271d757f8e2dd972ddcb45e7c90ad Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 13 Jul 2015 16:19:24 +0500 Subject: [PATCH 3/6] Update raw listener --- raw_socket_listener/listener.go | 97 +++++++++++++++++------------- raw_socket_listener/tcp_message.go | 29 ++++----- raw_socket_listener/tcp_packet.go | 50 +++++++-------- 3 files changed, 95 insertions(+), 81 deletions(-) diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index 9fd0c65..29534c2 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -1,4 +1,16 @@ -package raw_socket +/* +Package rawSocket provides traffic sniffier using RAW sockets. + +Capture traffic from socket using RAW_SOCKET's +http://en.wikipedia.org/wiki/Raw_socket + +RAW_SOCKET allow you listen for traffic on any port (e.g. sniffing) because they operate on IP level. + +Ports is TCP feature, same as flow control, reliable transmission and etc. + +This package implements own TCP layer: TCP packets is parsed using tcp_packet.go, and flow control is managed by tcp_message.go +*/ +package rawSocket import ( "bytes" @@ -8,41 +20,41 @@ import ( "strconv" ) -// Capture traffic from socket using RAW_SOCKET's -// http://en.wikipedia.org/wiki/Raw_socket -// -// RAW_SOCKET allow you listen for traffic on any port (e.g. sniffing) because they operate on IP level. -// Ports is TCP feature, same as flow control, reliable transmission and etc. -// Since we can't use default TCP libraries RAWTCPLitener implements own TCP layer -// TCP packets is parsed using tcp_packet.go, and flow control is managed by tcp_message.go +// Listener handle traffic capture type Listener struct { - messages map[string]*TCPMessage // buffer of TCPMessages waiting to be send + // buffer of TCPMessages waiting to be send + messages map[string]*TCPMessage // Expect: 100-continue request is send in 2 tcp messages // We store ACK aliases to merge this packets together - ack_aliases map[uint32]uint32 - seq_with_data map[uint32]uint32 + ackAliases map[uint32]uint32 + // To get ACK of second message we need to compute its Seq and wait for them message + seqWithData map[uint32]uint32 - c_packets chan *TCPPacket - c_messages chan *TCPMessage // Messages ready to be send to client + // Messages ready to be send to client + packetsChan chan *TCPPacket - c_del_message chan *TCPMessage // Used for notifications about completed or expired messages + // Messages ready to be send to client + messagesChan chan *TCPMessage + + // Used for notifications about completed or expired messages + messageDelChan chan *TCPMessage addr string // IP to listen port int // Port to listen } -// RAWTCPListen creates a listener to capture traffic from RAW_SOCKET +// NewListener creates and initializes new Listener object func NewListener(addr string, port string) (rawListener *Listener) { rawListener = &Listener{} - rawListener.c_packets = make(chan *TCPPacket, 10000) - rawListener.c_messages = make(chan *TCPMessage, 10000) - rawListener.c_del_message = make(chan *TCPMessage, 10000) + rawListener.packetsChan = make(chan *TCPPacket, 10000) + rawListener.messagesChan = make(chan *TCPMessage, 10000) + rawListener.messageDelChan = make(chan *TCPMessage, 10000) rawListener.messages = make(map[string]*TCPMessage) - rawListener.ack_aliases = make(map[uint32]uint32) - rawListener.seq_with_data = make(map[uint32]uint32) + rawListener.ackAliases = make(map[uint32]uint32) + rawListener.seqWithData = make(map[uint32]uint32) rawListener.addr = addr rawListener.port, _ = strconv.Atoi(port) @@ -57,13 +69,14 @@ func (t *Listener) listen() { for { select { // If message ready for deletion it means that its also complete or expired by timeout - case message := <-t.c_del_message: - t.c_messages <- message - delete(t.ack_aliases, message.Ack) + case message := <-t.messageDelChan: + log.Println("MESSAGE") + t.messagesChan <- message + delete(t.ackAliases, message.Ack) delete(t.messages, message.ID) // We need to use channels to process each packet to avoid data races - case packet := <-t.c_packets: + case packet := <-t.packetsChan: t.processTCPPacket(packet) } } @@ -94,18 +107,20 @@ func (t *Listener) readRAWSocket() { } func (t *Listener) parsePacket(addr net.Addr, buf []byte) { + log.Println("PACKET") + if t.isIncomingDataPacket(buf) { - t.c_packets <- ParseTCPPacket(addr, buf) + t.packetsChan <- ParseTCPPacket(addr, buf) } } func (t *Listener) isIncomingDataPacket(buf []byte) bool { // To avoid full packet parsing every time, we manually parsing values needed for packet filtering // http://en.wikipedia.org/wiki/Transmission_Control_Protocol - dest_port := binary.BigEndian.Uint16(buf[2:4]) + destPort := binary.BigEndian.Uint16(buf[2:4]) // Because RAW_SOCKET can't be bound to port, we have to control it by ourself - if int(dest_port) == t.port { + if int(destPort) == t.port { // Get the 'data offset' (size of the TCP header in 32-bit words) dataOffset := (buf[12] & 0xF0) >> 4 @@ -131,29 +146,27 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) { var message *TCPMessage - parent_message_ack, parent_ok := t.seq_with_data[packet.Seq] - if parent_ok { - t.ack_aliases[packet.Ack] = parent_message_ack - delete(t.seq_with_data, packet.Seq) + if parentAck, ok := t.seqWithData[packet.Seq]; ok { + t.ackAliases[packet.Ack] = parentAck + delete(t.seqWithData, packet.Seq) } - ack_alias, alias_ok := t.ack_aliases[packet.Ack] - if alias_ok { - packet.Ack = ack_alias + if alias, ok := t.ackAliases[packet.Ack]; ok { + packet.Ack = alias } - m_id := packet.Addr.String() + strconv.Itoa(int(packet.SrcPort)) + strconv.Itoa(int(packet.Ack)) - message, ok := t.messages[m_id] + mID := packet.Addr.String() + strconv.Itoa(int(packet.SrcPort)) + strconv.Itoa(int(packet.Ack)) + message, ok := t.messages[mID] if !ok { - // We sending c_del_message channel, so message object can communicate with Listener and notify it if message completed - message = NewTCPMessage(m_id, t.c_del_message, packet.Ack) - t.messages[m_id] = message + // We sending messageDelChan channel, so message object can communicate with Listener and notify it if message completed + message = NewTCPMessage(mID, t.messageDelChan, packet.Ack) + t.messages[mID] = message } if bytes.Equal(packet.Data[0:4], bPOST) { if bytes.Equal(packet.Data[len(packet.Data)-24:len(packet.Data)-4], bExpect100ContinueCheck) { - t.seq_with_data[packet.Seq+uint32(len(packet.Data))] = packet.Ack + t.seqWithData[packet.Seq+uint32(len(packet.Data))] = packet.Ack // Removing `Expect: 100-continue` header packet.Data = append(packet.Data[:len(packet.Data)-24], packet.Data[len(packet.Data)-2:]...) @@ -161,10 +174,10 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) { } // Adding packet to message - message.c_packets <- packet + message.packetsChan <- packet } // Receive TCP messages from the listener channel func (t *Listener) Receive() *TCPMessage { - return <-t.c_messages + return <-t.messagesChan } diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index 32fdaae..ddfb0f2 100644 --- a/raw_socket_listener/tcp_message.go +++ b/raw_socket_listener/tcp_message.go @@ -1,4 +1,4 @@ -package raw_socket +package rawSocket import ( "log" @@ -6,7 +6,8 @@ import ( "time" ) -const MSG_EXPIRE = 2000 * time.Millisecond +// MsgExpire specify period that message should wait before it considered as finished +const MsgExpire = 2000 * time.Millisecond // TCPMessage ensure that all TCP packets for given request is received, and processed in right sequence // Its needed because all TCP message can be fragmented or re-transmitted @@ -21,20 +22,20 @@ type TCPMessage struct { timer *time.Timer // Used for expire check - c_packets chan *TCPPacket + packetsChan chan *TCPPacket - c_del_message chan *TCPMessage + delChan chan *TCPMessage } // NewTCPMessage pointer created from a Acknowledgment number and a channel of messages readuy to be deleted -func NewTCPMessage(ID string, c_del chan *TCPMessage, Ack uint32) (msg *TCPMessage) { +func NewTCPMessage(ID string, delChan chan *TCPMessage, Ack uint32) (msg *TCPMessage) { msg = &TCPMessage{ID: ID, Ack: Ack} - msg.c_packets = make(chan *TCPPacket) - msg.c_del_message = c_del // used for notifying that message completed or expired + msg.packetsChan = make(chan *TCPPacket) + msg.delChan = delChan // used for notifying that message completed or expired // Every time we receive packet we reset this timer - msg.timer = time.AfterFunc(MSG_EXPIRE, msg.Timeout) + msg.timer = time.AfterFunc(MsgExpire, msg.Timeout) go msg.listen() @@ -44,7 +45,7 @@ func NewTCPMessage(ID string, c_del chan *TCPMessage, Ack uint32) (msg *TCPMessa func (t *TCPMessage) listen() { for { select { - case packet, more := <-t.c_packets: + case packet, more := <-t.packetsChan: if more { t.AddPacket(packet) } else { @@ -60,21 +61,21 @@ func (t *TCPMessage) Timeout() { select { // In some cases Timeout can be called multiple times (do not know how yet) // Ensure that we did not close channel 2 times - case packet, ok := <-t.c_packets: + case packet, ok := <-t.packetsChan: if ok { t.AddPacket(packet) } else { return } default: - close(t.c_packets) - t.c_del_message <- t // Notify RAWListener that message is ready to be send to replay server + close(t.packetsChan) + t.delChan <- t // Notify RAWListener that message is ready to be send to replay server } } // Bytes sorts packets in right orders and return message content func (t *TCPMessage) Bytes() (output []byte) { - sort.Sort(BySeq(t.packets)) + sort.Sort(sortBySeq(t.packets)) for _, v := range t.packets { output = append(output, v.Data...) @@ -102,5 +103,5 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) { } // Reset message timeout timer - t.timer.Reset(MSG_EXPIRE) + t.timer.Reset(MsgExpire) } diff --git a/raw_socket_listener/tcp_packet.go b/raw_socket_listener/tcp_packet.go index 4d3f4e6..3ebcf88 100644 --- a/raw_socket_listener/tcp_packet.go +++ b/raw_socket_listener/tcp_packet.go @@ -1,4 +1,4 @@ -package raw_socket +package rawSocket import ( "encoding/binary" @@ -9,19 +9,18 @@ import ( // TCP Flags const ( - TCP_FIN = 1 << iota - TCP_SYN - TCP_RST - TCP_PSH - TCP_ACK - TCP_URG - TCP_ECE - TCP_CWR - TCP_NS + fFIN = 1 << iota + fSYN + fRST + fPSH + fACK + fURG + fECE + fCWR + fNS ) -// Simple TCP packet parser -// +// TCPPacket provides tcp packet parser // Packet structure: http://en.wikipedia.org/wiki/Transmission_Control_Protocol type TCPPacket struct { SrcPort uint16 @@ -39,6 +38,7 @@ type TCPPacket struct { Addr net.Addr } +// ParseTCPPacket takes address and tcp payload and returns parsed TCPPacket func ParseTCPPacket(addr net.Addr, b []byte) (p *TCPPacket) { p = &TCPPacket{Data: b} p.ParseBasic() @@ -76,15 +76,15 @@ func (t *TCPPacket) String() string { "Acknowledgment:" + strconv.Itoa(int(t.Ack)), "Header len:" + strconv.Itoa(int(t.DataOffset)), - "Flag ns:" + strconv.FormatBool(t.Flags&TCP_NS != 0), - "Flag crw:" + strconv.FormatBool(t.Flags&TCP_CWR != 0), - "Flag ece:" + strconv.FormatBool(t.Flags&TCP_ECE != 0), - "Flag urg:" + strconv.FormatBool(t.Flags&TCP_URG != 0), - "Flag ack:" + strconv.FormatBool(t.Flags&TCP_ACK != 0), - "Flag psh:" + strconv.FormatBool(t.Flags&TCP_PSH != 0), - "Flag rst:" + strconv.FormatBool(t.Flags&TCP_RST != 0), - "Flag syn:" + strconv.FormatBool(t.Flags&TCP_SYN != 0), - "Flag fin:" + strconv.FormatBool(t.Flags&TCP_FIN != 0), + "Flag ns:" + strconv.FormatBool(t.Flags&fNS != 0), + "Flag crw:" + strconv.FormatBool(t.Flags&fCWR != 0), + "Flag ece:" + strconv.FormatBool(t.Flags&fECE != 0), + "Flag urg:" + strconv.FormatBool(t.Flags&fURG != 0), + "Flag ack:" + strconv.FormatBool(t.Flags&fACK != 0), + "Flag psh:" + strconv.FormatBool(t.Flags&fPSH != 0), + "Flag rst:" + strconv.FormatBool(t.Flags&fRST != 0), + "Flag syn:" + strconv.FormatBool(t.Flags&fSYN != 0), + "Flag fin:" + strconv.FormatBool(t.Flags&fFIN != 0), "Window size:" + strconv.Itoa(int(t.Window)), "Checksum:" + strconv.Itoa(int(t.Checksum)), @@ -94,8 +94,8 @@ func (t *TCPPacket) String() string { }, "\n") } -type BySeq []*TCPPacket +type sortBySeq []*TCPPacket -func (a BySeq) Len() int { return len(a) } -func (a BySeq) Swap(i, j int) { a[i], a[j] = a[j], a[i] } -func (a BySeq) Less(i, j int) bool { return a[i].Seq < a[j].Seq } +func (a sortBySeq) Len() int { return len(a) } +func (a sortBySeq) Swap(i, j int) { a[i], a[j] = a[j], a[i] } +func (a sortBySeq) Less(i, j int) bool { return a[i].Seq < a[j].Seq } From 6158ee30255ce6a651dd6f1712d87b4ee66b2f03 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 13 Jul 2015 16:20:57 +0500 Subject: [PATCH 4/6] Remove debug --- raw_socket_listener/listener.go | 3 --- 1 file changed, 3 deletions(-) diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index 29534c2..280e288 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -70,7 +70,6 @@ func (t *Listener) listen() { select { // If message ready for deletion it means that its also complete or expired by timeout case message := <-t.messageDelChan: - log.Println("MESSAGE") t.messagesChan <- message delete(t.ackAliases, message.Ack) delete(t.messages, message.ID) @@ -107,8 +106,6 @@ func (t *Listener) readRAWSocket() { } func (t *Listener) parsePacket(addr net.Addr, buf []byte) { - log.Println("PACKET") - if t.isIncomingDataPacket(buf) { t.packetsChan <- ParseTCPPacket(addr, buf) } From 28540f43a11bfe0932fb1ab49a749d5bb270a9b6 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 13 Jul 2015 16:56:52 +0500 Subject: [PATCH 5/6] New portion of changes --- output_http.go | 32 +++++++++++++++++++------------- output_http_test.go | 4 ++-- output_tcp.go | 5 +++++ plugins.go | 23 +++++++++++++---------- settings.go | 9 +++++++-- test_input.go | 7 +++++++ test_output.go | 2 ++ 7 files changed, 55 insertions(+), 27 deletions(-) diff --git a/output_http.go b/output_http.go index e2a6c96..eb3e925 100644 --- a/output_http.go +++ b/output_http.go @@ -7,8 +7,9 @@ import ( "time" ) -const InitialDynamicWorkers = 10 +const initialDynamicWorkers = 10 +// HTTPOutputConfig struct for holding http output configuration type HTTPOutputConfig struct { redirectLimit int @@ -20,6 +21,9 @@ type HTTPOutputConfig struct { Debug bool } +// 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` 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 @@ -39,6 +43,8 @@ type HTTPOutput struct { elasticSearch *ESPlugin } +// NewHTTPOutput constructor for HTTPOutput +// Initialize workers func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer { o := new(HTTPOutput) @@ -55,7 +61,7 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer { // Initial workers count if o.config.workers == 0 { - o.needWorker <- InitialDynamicWorkers + o.needWorker <- initialDynamicWorkers } else { o.needWorker <- o.config.workers } @@ -65,16 +71,16 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer { o.elasticSearch.Init(o.config.elasticSearch) } - go o.WorkerMaster() + go o.workerMaster() return o } -func (o *HTTPOutput) WorkerMaster() { +func (o *HTTPOutput) workerMaster() { for { - new_workers := <-o.needWorker - for i := 0; i < new_workers; i++ { - go o.Worker() + newWorkers := <-o.needWorker + for i := 0; i < newWorkers; i++ { + go o.startWorker() } // Disable dynamic scaling if workers poll fixed size @@ -84,13 +90,13 @@ func (o *HTTPOutput) WorkerMaster() { } } -func (o *HTTPOutput) Worker() { +func (o *HTTPOutput) startWorker() { client := NewHTTPClient(o.address, &HTTPClientConfig{ FollowRedirects: o.config.redirectLimit, Debug: o.config.Debug, }) - death_count := 0 + deathCount := 0 atomic.AddInt64(&o.activeWorkers, 1) @@ -98,19 +104,19 @@ func (o *HTTPOutput) Worker() { select { case data := <-o.queue: o.sendRequest(client, data) - death_count = 0 + deathCount = 0 case <-time.After(time.Millisecond * 100): // When dynamic scaling enabled workers die after 2s of inactivity if o.config.workers == 0 { - death_count += 1 + deathCount++ } else { continue } - if death_count > 20 { + if deathCount > 20 { workersCount := atomic.LoadInt64(&o.activeWorkers) - // At least 1 worker should be alive + // At least 1 startWorker should be alive if workersCount != 1 { atomic.AddInt64(&o.activeWorkers, -1) return diff --git a/output_http_test.go b/output_http_test.go index 7cf8c24..16f8422 100644 --- a/output_http_test.go +++ b/output_http_test.go @@ -86,10 +86,10 @@ func TestOutputHTTPSSL(t *testing.T) { })) input := NewTestInput() - http_output := NewHTTPOutput(server.URL, &HTTPOutputConfig{}) + output := NewHTTPOutput(server.URL, &HTTPOutputConfig{}) Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{http_output} + Plugins.Outputs = []io.Writer{output} go Start(quit) diff --git a/output_tcp.go b/output_tcp.go index f6f0e9d..dd7cdc4 100644 --- a/output_tcp.go +++ b/output_tcp.go @@ -9,6 +9,9 @@ import ( "time" ) +// TCPOutput used for sending raw tcp payloads +// Currently used for internal communication between listener and replay server +// Can be used for transfering binary payloads like protocol buffers type TCPOutput struct { address string limit int @@ -16,6 +19,8 @@ type TCPOutput struct { bufStats *GorStat } +// NewTCPOutput constructor for TCPOutput +// Initialize 10 workers which hold keep-alive connection func NewTCPOutput(address string) io.Writer { o := new(TCPOutput) diff --git a/plugins.go b/plugins.go index 88134ef..cb681ae 100644 --- a/plugins.go +++ b/plugins.go @@ -6,23 +6,25 @@ import ( "strings" ) +// InOutPlugins struct for holding references to plugins type InOutPlugins struct { Inputs []io.Reader Outputs []io.Writer } -type ReaderOrWriter interface{} - -var Plugins *InOutPlugins = new(InOutPlugins) +// Plugins holds all the plugin objects +var Plugins *InOutPlugins +// extractLimitOptions detects if plugin get called with limiter support +// Returns address and limit func extractLimitOptions(options string) (string, string) { split := strings.Split(options, "|") if len(split) > 1 { return split[0], split[1] - } else { - return split[0], "" } + + return split[0], "" } // Automatically detects type of plugin and initialize it @@ -45,23 +47,24 @@ func registerPlugin(constructor interface{}, options ...interface{}) { // Calling our constructor with list of given options plugin := vc.Call(vo)[0].Interface() - plugin_wrapper := plugin + pluginWrapper := plugin if limit != "" { - plugin_wrapper = NewLimiter(plugin, limit) + pluginWrapper = NewLimiter(plugin, limit) } else { - plugin_wrapper = plugin + pluginWrapper = plugin } if _, ok := plugin.(io.Reader); ok { - Plugins.Inputs = append(Plugins.Inputs, plugin_wrapper.(io.Reader)) + Plugins.Inputs = append(Plugins.Inputs, pluginWrapper.(io.Reader)) } if _, ok := plugin.(io.Writer); ok { - Plugins.Outputs = append(Plugins.Outputs, plugin_wrapper.(io.Writer)) + Plugins.Outputs = append(Plugins.Outputs, pluginWrapper.(io.Writer)) } } +// InitPlugins specify and initialize all available plugins func InitPlugins() { for _, options := range Settings.inputDummy { registerPlugin(NewDummyInput, options) diff --git a/settings.go b/settings.go index 8c345e7..a08a9c1 100644 --- a/settings.go +++ b/settings.go @@ -8,21 +8,24 @@ import ( ) const ( + // VERSION specifies Gor current version VERSION = "0.9.8" ) -// Allows to specify multiple flags with same name and collects all values to array +// MultiOption allows to specify multiple flags with same name and collects all values into array type MultiOption []string func (h *MultiOption) String() string { return fmt.Sprint(*h) } +// Set gets called multiple times for each flag with same name func (h *MultiOption) Set(value string) error { *h = append(*h, value) return nil } +// AppSettings is the struct of main configuration type AppSettings struct { verbose bool debug bool @@ -49,7 +52,8 @@ type AppSettings struct { modifierConfig HTTPModifierConfig } -var Settings AppSettings = AppSettings{} +// Settings holds Gor configuration +var Settings AppSettings func usage() { fmt.Printf("Gor is a simple http traffic replication tool written in Go. Its main goal is to replay traffic from production servers to staging and dev environments.\nProject page: https://github.com/buger/gor\nAuthor: leonsbox@gmail.com\nCurrent Version: %s\n\n", VERSION) @@ -113,6 +117,7 @@ func init() { flag.Var(&Settings.modifierConfig.paramHashFilters, "http-param-limiter", "Takes a fraction of requests, consistently taking or rejecting a request based on the FNV32-1A hash of a specific GET param:\n\t gor --input-raw :8080 --output-http staging.com --http-param-limiter user_id:25%") } +// Debug gets called only if --verbose flag specified func Debug(args ...interface{}) { if Settings.verbose { fmt.Print("[DEBUG] ") diff --git a/test_input.go b/test_input.go index 8f62344..a9a359a 100644 --- a/test_input.go +++ b/test_input.go @@ -5,10 +5,12 @@ import ( "encoding/base64" ) +// TestInput used for testing purpose, it allows emitting requests on demand type TestInput struct { data chan []byte } +// NewTestInput constructor for TestInput func NewTestInput() (i *TestInput) { i = new(TestInput) i.data = make(chan []byte, 100) @@ -23,18 +25,22 @@ func (i *TestInput) Read(data []byte) (int, error) { return len(buf), nil } +// EmitGET emits GET request without headers func (i *TestInput) EmitGET() { i.data <- []byte("GET / HTTP/1.1\r\n\r\n") } +// EmitPOST emits POST request with Content-Length func (i *TestInput) EmitPOST() { i.data <- []byte("POST /pub/WWW/ HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") } +// EmitChunkedPOST emits POST request with `Transfer-Encoding: chunked` and chunked body func (i *TestInput) EmitChunkedPOST() { i.data <- []byte("POST /pub/WWW/ 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") } +// EmitLargePOST emits POST request with large payload (5mb) func (i *TestInput) EmitLargePOST() { size := 5 * 1024 * 1024 // 5 MB rb := make([]byte, size) @@ -45,6 +51,7 @@ func (i *TestInput) EmitLargePOST() { i.data <- []byte("POST / HTTP/1.1\nHost: www.w3.org\nContent-Length:5242880\r\n\r\n" + rs) } +// EmitOPTIONS emits OPTIONS request, similar to GET func (i *TestInput) EmitOPTIONS() { i.data <- []byte("OPTIONS / HTTP/1.1\nHost: www.w3.org\r\n\r\n") } diff --git a/test_output.go b/test_output.go index 1d34da0..e76fbb6 100644 --- a/test_output.go +++ b/test_output.go @@ -2,10 +2,12 @@ package main type writeCallback func(data []byte) +// TestOutput used in testing to intercept any output into callback type TestOutput struct { cb writeCallback } +// NewTestOutput constructor for TestOutput, accepts callback which get called on each incoming Write func NewTestOutput(cb writeCallback) (i *TestOutput) { i = new(TestOutput) i.cb = cb From 125d9db83410c354adf72228f5936454c11022bd Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Fri, 17 Jul 2015 11:30:43 +0500 Subject: [PATCH 6/6] Next batch --- .travis.yml | 4 ++-- emitter.go | 3 ++- http_client.go | 2 +- http_client_test.go | 24 ++++++++++----------- http_modifier_settings.go | 6 ++++-- http_modifier_test.go | 22 ++++++++++---------- input_dummy.go | 2 ++ input_file.go | 6 ++++-- input_http.go | 2 ++ input_raw.go | 2 ++ input_raw_test.go | 44 +++++++++++++++++++-------------------- input_tcp.go | 6 +++--- input_tcp_test.go | 6 +++--- limiter.go | 3 +++ output_dummy.go | 2 ++ output_file.go | 8 +++++-- plugins.go | 2 +- 17 files changed, 82 insertions(+), 62 deletions(-) diff --git a/.travis.yml b/.travis.yml index a236b6a..7f2d291 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,3 +1,3 @@ -language: go +language: go go: 1.4.2 -script: sudo -E bash -c "source /etc/profile && eval '$(gimme 1.4)' && export GOPATH=$HOME/gopath:$GOPATH && go get && GORACE='halt_on_error=1' go test -race -v" +script: sudo -E bash -c "source /etc/profile && eval '$(gimme 1.4.2)' && export GOPATH=$HOME/gopath:$GOPATH && go get && GORACE='halt_on_error=1' go test -v" \ No newline at end of file diff --git a/emitter.go b/emitter.go index 2181331..1f7caa0 100644 --- a/emitter.go +++ b/emitter.go @@ -5,6 +5,7 @@ import ( "time" ) +// Start initialize loop for sending data from inputs to outputs func Start(stop chan int) { for _, in := range Plugins.Inputs { go CopyMulty(in, Plugins.Outputs...) @@ -19,7 +20,7 @@ func Start(stop chan int) { } } -// Copy from 1 reader to multiple writers +// CopyMulty copies from 1 reader to multiple writers func CopyMulty(src io.Reader, writers ...io.Writer) (err error) { buf := make([]byte, 5*1024*1024) wIndex := 0 diff --git a/http_client.go b/http_client.go index c54e6a8..f9ad04d 100644 --- a/http_client.go +++ b/http_client.go @@ -145,7 +145,7 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { // 3xx requests if status[0] == '3' { - c.redirectsCount += 1 + c.redirectsCount++ location := proto.Header(payload, []byte("Location")) redirectPayload := []byte("GET " + string(location) + " HTTP/1.1\r\n\r\n") diff --git a/http_client_test.go b/http_client_test.go index a5b238b..feb0326 100644 --- a/http_client_test.go +++ b/http_client_test.go @@ -132,7 +132,7 @@ func TestHTTPClientHTTPSSend(t *testing.T) { func TestHTTPClientServerInstantDisconnect(t *testing.T) { wg := new(sync.WaitGroup) - GET_payload := []byte("GET / HTTP/1.1\r\n\r\n") + GETPayload := []byte("GET / HTTP/1.1\r\n\r\n") ln, _ := net.Listen("tcp", ":0") @@ -148,8 +148,8 @@ func TestHTTPClientServerInstantDisconnect(t *testing.T) { client := NewHTTPClient(ln.Addr().String(), &HTTPClientConfig{}) wg.Add(2) - client.Send(GET_payload) - client.Send(GET_payload) + client.Send(GETPayload) + client.Send(GETPayload) wg.Wait() } @@ -157,7 +157,7 @@ func TestHTTPClientServerInstantDisconnect(t *testing.T) { func TestHTTPClientServerNoKeepAlive(t *testing.T) { wg := new(sync.WaitGroup) - GET_payload := []byte("GET / HTTP/1.1\r\n\r\n") + GETPayload := []byte("GET / HTTP/1.1\r\n\r\n") ln, _ := net.Listen("tcp", ":0") @@ -186,8 +186,8 @@ func TestHTTPClientServerNoKeepAlive(t *testing.T) { client := NewHTTPClient(ln.Addr().String(), &HTTPClientConfig{}) wg.Add(2) - client.Send(GET_payload) - client.Send(GET_payload) + client.Send(GETPayload) + client.Send(GETPayload) wg.Wait() } @@ -195,7 +195,7 @@ func TestHTTPClientServerNoKeepAlive(t *testing.T) { func TestHTTPClientRedirect(t *testing.T) { wg := new(sync.WaitGroup) - GET_payload := []byte("GET / HTTP/1.1\r\n\r\n") + GETPayload := []byte("GET / HTTP/1.1\r\n\r\n") server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -210,7 +210,7 @@ func TestHTTPClientRedirect(t *testing.T) { // Should do 2 queries wg.Add(2) - client.Send(GET_payload) + client.Send(GETPayload) wg.Wait() } @@ -218,7 +218,7 @@ func TestHTTPClientRedirect(t *testing.T) { func TestHTTPClientRedirectLimit(t *testing.T) { wg := new(sync.WaitGroup) - GET_payload := []byte("GET / HTTP/1.1\r\n\r\n") + GETPayload := []byte("GET / HTTP/1.1\r\n\r\n") server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -241,7 +241,7 @@ func TestHTTPClientRedirectLimit(t *testing.T) { // Have 3 redirects + 1 GET, but should do only 2 redirects + GET wg.Add(3) - client.Send(GET_payload) + client.Send(GETPayload) wg.Wait() } @@ -249,7 +249,7 @@ func TestHTTPClientRedirectLimit(t *testing.T) { func TestHTTPClientHandleHTTP10(t *testing.T) { wg := new(sync.WaitGroup) - GET_payload := []byte("GET http://foobar.com/path HTTP/1.0\r\n\r\n") + 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) { @@ -263,7 +263,7 @@ func TestHTTPClientHandleHTTP10(t *testing.T) { client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true}) wg.Add(1) - client.Send(GET_payload) + client.Send(GETPayload) wg.Wait() } diff --git a/http_modifier_settings.go b/http_modifier_settings.go index 854b11a..1a7323a 100644 --- a/http_modifier_settings.go +++ b/http_modifier_settings.go @@ -8,6 +8,7 @@ import ( "strings" ) +// HTTPModifierConfig holds configuration options for built-in traffic modifier type HTTPModifierConfig struct { urlNegativeRegexp HTTPUrlRegexp urlRegexp HTTPUrlRegexp @@ -29,6 +30,7 @@ type headerFilter struct { regexp *regexp.Regexp } +// HTTPHeaderFilters holds list of headers and their regexps type HTTPHeaderFilters []headerFilter func (h *HTTPHeaderFilters) String() string { @@ -67,7 +69,7 @@ func (h *HTTPHashFilters) String() string { func (h *HTTPHashFilters) Set(value string) error { valArr := strings.SplitN(value, ":", 2) if len(valArr) < 2 { - return errors.New("need both header and value, colon-delimited (ex. user_id:50%).") + return errors.New("need both header and value, colon-delimited (ex. user_id:50%)") } f := hashFilter{name: []byte(valArr[0])} @@ -180,7 +182,7 @@ func (r *UrlRewriteMap) String() string { func (r *UrlRewriteMap) Set(value string) error { valArr := strings.SplitN(value, ":", 2) if len(valArr) < 2 { - return errors.New("need both src and target, colon-delimited (ex. /a:/b).") + return errors.New("need both src and target, colon-delimited (ex. /a:/b)") } regexp, err := regexp.Compile(valArr[0]) if err != nil { diff --git a/http_modifier_test.go b/http_modifier_test.go index 40e5ec4..b4e9348 100644 --- a/http_modifier_test.go +++ b/http_modifier_test.go @@ -40,7 +40,7 @@ func TestHTTPModifierHeaderFilters(t *testing.T) { } func TestHTTPModifierURLRewrite(t *testing.T) { - var url, new_url []byte + var url, newURL []byte rewrites := UrlRewriteMap{} @@ -58,13 +58,13 @@ func TestHTTPModifierURLRewrite(t *testing.T) { }) url = []byte("/v1/user/joe/ping") - if new_url = proto.Path(modifier.Rewrite(payload(url))); bytes.Equal(new_url, url) { - t.Error("Request url should have been rewritten, wasn't", string(new_url)) + if newURL = proto.Path(modifier.Rewrite(payload(url))); bytes.Equal(newURL, url) { + t.Error("Request url should have been rewritten, wasn't", string(newURL)) } url = []byte("/v1/user/ping") - if new_url = proto.Path(modifier.Rewrite(payload(url))); !bytes.Equal(new_url, url) { - t.Error("Request url should have been rewritten, wasn't", string(new_url)) + if newURL = proto.Path(modifier.Rewrite(payload(url))); !bytes.Equal(newURL, url) { + t.Error("Request url should have been rewritten, wasn't", string(newURL)) } } @@ -128,9 +128,9 @@ func TestHTTPModifierHeaders(t *testing.T) { }) payload := []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - new_payload := []byte("POST /post HTTP/1.1\r\nHeader1: 1\r\nContent-Length: 7\r\nHost: localhost\r\n\r\na=1&b=2") + newPayload := []byte("POST /post HTTP/1.1\r\nHeader1: 1\r\nContent-Length: 7\r\nHost: localhost\r\n\r\na=1&b=2") - if payload = modifier.Rewrite(payload); !bytes.Equal(payload, new_payload) { + if payload = modifier.Rewrite(payload); !bytes.Equal(payload, newPayload) { t.Error("Should update request headers", string(payload)) } } @@ -196,9 +196,9 @@ func TestHTTPModifierSetHeader(t *testing.T) { }) payload := []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - payload_after := []byte("POST /post HTTP/1.1\r\nUser-Agent: Gor\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") + payloadAfter := []byte("POST /post HTTP/1.1\r\nUser-Agent: Gor\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - if payload = modifier.Rewrite(payload); !bytes.Equal(payload_after, payload) { + if payload = modifier.Rewrite(payload); !bytes.Equal(payloadAfter, payload) { t.Error("Should add new header", string(payload)) } } @@ -212,9 +212,9 @@ func TestHTTPModifierSetParam(t *testing.T) { }) payload := []byte("POST /post?api_key=1234 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - payload_after := []byte("POST /post?api_key=1 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") + payloadAfter := []byte("POST /post?api_key=1 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - if payload = modifier.Rewrite(payload); !bytes.Equal(payload_after, payload) { + if payload = modifier.Rewrite(payload); !bytes.Equal(payloadAfter, payload) { t.Error("Should override param", string(payload)) } } diff --git a/input_dummy.go b/input_dummy.go index a2a87ac..7f6a7f8 100644 --- a/input_dummy.go +++ b/input_dummy.go @@ -4,10 +4,12 @@ import ( "time" ) +// DummyInput used for debugging. It generate 1 "GET /"" request per second. type DummyInput struct { data chan []byte } +// NewDummyInput constructor for DummyInput func NewDummyInput(options string) (di *DummyInput) { di = new(DummyInput) di.data = make(chan []byte) diff --git a/input_file.go b/input_file.go index 043ca3e..60c93ec 100644 --- a/input_file.go +++ b/input_file.go @@ -7,6 +7,7 @@ import ( "time" ) +// FileInput can read requests generated by FileOutput type FileInput struct { data chan []byte path string @@ -14,19 +15,20 @@ type FileInput struct { speedFactor float64 } +// NewFileInput constructor for FileInput. Accepts file path as argument. func NewFileInput(path string) (i *FileInput) { i = new(FileInput) i.data = make(chan []byte) i.path = path i.speedFactor = 1 - i.Init(path) + i.init(path) go i.emit() return } -func (i *FileInput) Init(path string) { +func (i *FileInput) init(path string) { file, err := os.Open(path) if err != nil { diff --git a/input_http.go b/input_http.go index 0ed16eb..5150c4e 100644 --- a/input_http.go +++ b/input_http.go @@ -7,12 +7,14 @@ import ( "net/http/httputil" ) +// HTTPInput used for sending requests to Gor via http type HTTPInput struct { data chan []byte address string listener net.Listener } +// NewHTTPInput constructor for HTTPInput. Accepts address with port which he will listen on. func NewHTTPInput(address string) (i *HTTPInput) { i = new(HTTPInput) i.data = make(chan []byte) diff --git a/input_raw.go b/input_raw.go index 3a6a960..93b007d 100644 --- a/input_raw.go +++ b/input_raw.go @@ -7,11 +7,13 @@ import ( "strings" ) +// RAWInput used for intercepting traffic for given address type RAWInput struct { data chan []byte address string } +// NewRAWInput constructor for RAWInput. Accepts address with port as argument. func NewRAWInput(address string) (i *RAWInput) { i = new(RAWInput) i.data = make(chan []byte) diff --git a/input_raw_test.go b/input_raw_test.go index 6262667..7380fad 100644 --- a/input_raw_test.go +++ b/input_raw_test.go @@ -49,7 +49,7 @@ func TestInputRAW100Expect(t *testing.T) { wg := new(sync.WaitGroup) quit := make(chan int) - file_content, _ := ioutil.ReadFile("README.md") + fileContent, _ := ioutil.ReadFile("README.md") // Origing and Replay server initialization origin := startHTTP(func(req *http.Request) { @@ -59,12 +59,12 @@ func TestInputRAW100Expect(t *testing.T) { wg.Done() }) - origin_address := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1) + originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(origin_address) + input := NewRAWInput(originAddr) // We will use it to get content of raw HTTP request - test_output := NewTestOutput(func(data []byte) { + testOutput := NewTestOutput(func(data []byte) { if strings.Contains(string(data), "Expect: 100-continue") { t.Error("Should not contain 100-continue header") } @@ -75,24 +75,24 @@ func TestInputRAW100Expect(t *testing.T) { defer req.Body.Close() body, _ := ioutil.ReadAll(req.Body) - if !bytes.Equal(body, file_content) { + if !bytes.Equal(body, fileContent) { buf, _ := httputil.DumpRequest(req, true) t.Error("Wrong POST body:", string(buf)) } wg.Done() }) - replay_address := listener.Addr().String() + replayAddr := listener.Addr().String() - http_output := NewHTTPOutput(replay_address, &HTTPOutputConfig{}) + httpOutput := NewHTTPOutput(replayAddr, &HTTPOutputConfig{}) Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{test_output, http_output} + Plugins.Outputs = []io.Writer{testOutput, httpOutput} go Start(quit) wg.Add(3) - curl := exec.Command("curl", "http://"+origin_address, "--data-binary", "@README.md") + curl := exec.Command("curl", "http://"+originAddr, "--data-binary", "@README.md") err := curl.Run() if err != nil { log.Fatal(err) @@ -106,7 +106,7 @@ func TestInputRAWChunkedEncoding(t *testing.T) { wg := new(sync.WaitGroup) quit := make(chan int) - file_content, _ := ioutil.ReadFile("README.md") + fileContent, _ := ioutil.ReadFile("README.md") // Origing and Replay server initialization origin := startHTTP(func(req *http.Request) { @@ -116,33 +116,33 @@ func TestInputRAWChunkedEncoding(t *testing.T) { wg.Done() }) - origin_address := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1) + originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(origin_address) + input := NewRAWInput(originAddr) listener := startHTTP(func(req *http.Request) { defer req.Body.Close() body, _ := ioutil.ReadAll(req.Body) - if !bytes.Equal(body, file_content) { + if !bytes.Equal(body, fileContent) { buf, _ := httputil.DumpRequest(req, true) t.Error("Wrong POST body:", string(buf)) } wg.Done() }) - replay_address := listener.Addr().String() + replayAddr := listener.Addr().String() - http_output := NewHTTPOutput(replay_address, &HTTPOutputConfig{Debug: true}) + httpOutput := NewHTTPOutput(replayAddr, &HTTPOutputConfig{Debug: true}) Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{http_output} + Plugins.Outputs = []io.Writer{httpOutput} go Start(quit) wg.Add(2) - curl := exec.Command("curl", "http://"+origin_address, "--header", "Transfer-Encoding: chunked", "--data-binary", "@README.md") + curl := exec.Command("curl", "http://"+originAddr, "--header", "Transfer-Encoding: chunked", "--data-binary", "@README.md") err := curl.Run() if err != nil { log.Fatal(err) @@ -175,9 +175,9 @@ func TestInputRAWLargePayload(t *testing.T) { wg.Done() }) - origin_address := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1) + originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(origin_address) + input := NewRAWInput(originAddr) replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { req.Body = http.MaxBytesReader(w, req.Body, 1*1024*1024) @@ -193,15 +193,15 @@ func TestInputRAWLargePayload(t *testing.T) { })) defer replay.Close() - http_output := NewHTTPOutput(replay.URL, &HTTPOutputConfig{Debug: false}) + httpOutput := NewHTTPOutput(replay.URL, &HTTPOutputConfig{Debug: false}) Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{http_output} + Plugins.Outputs = []io.Writer{httpOutput} go Start(quit) wg.Add(2) - curl := exec.Command("curl", "http://"+origin_address, "--header", "Transfer-Encoding: chunked", "--data-binary", "@/tmp/large") + curl := exec.Command("curl", "http://"+originAddr, "--header", "Transfer-Encoding: chunked", "--data-binary", "@/tmp/large") err = curl.Run() if err != nil { log.Fatal("curl error:", err) diff --git a/input_tcp.go b/input_tcp.go index 27b7de6..9a2da90 100644 --- a/input_tcp.go +++ b/input_tcp.go @@ -9,15 +9,15 @@ import ( "os" ) -// Can be tested using nc tool: -// echo "asdad" | nc 127.0.0.1 27017 -// +// TCPInput used for internal communication +// It expected hex encoded data type TCPInput struct { data chan []byte address string listener net.Listener } +// NewTCPInput constructor for TCPInput, accepts address with port func NewTCPInput(address string) (i *TCPInput) { i = new(TCPInput) i.data = make(chan []byte) diff --git a/input_tcp_test.go b/input_tcp_test.go index 27c00fc..bbb0cc7 100644 --- a/input_tcp_test.go +++ b/input_tcp_test.go @@ -83,10 +83,10 @@ func BenchmarkTCPInput(b *testing.B) { for { data := <-dataChan - new_buf := make([]byte, len(data)+2) + buf := make([]byte, len(data)+2) data = append(data, []byte("ΒΆ")...) - copy(new_buf, data) - conn.Write(new_buf) + copy(buf, data) + conn.Write(buf) } }(conn) } diff --git a/limiter.go b/limiter.go index 3bd5acb..0e317a0 100644 --- a/limiter.go +++ b/limiter.go @@ -9,6 +9,7 @@ import ( "time" ) +// Limiter is a wrapper for input or output plugin which adds rate limiting type Limiter struct { plugin interface{} limit int @@ -30,6 +31,8 @@ func parseLimitOptions(options string) (limit int, isPercent bool) { return } +// NewLimiter constructor for Limiter, accepts plugin and options +// `options` allow to sprcify relatve or absolute limiting func NewLimiter(plugin interface{}, options string) io.ReadWriter { l := new(Limiter) l.limit, l.isPercent = parseLimitOptions(options) diff --git a/output_dummy.go b/output_dummy.go index f730c81..a033ebd 100644 --- a/output_dummy.go +++ b/output_dummy.go @@ -4,9 +4,11 @@ import ( "fmt" ) +// DummyOutput used for debugging, prints all incoming requests type DummyOutput struct { } +// NewDummyOutput constructor for DummyOutput func NewDummyOutput(options string) (di *DummyOutput) { di = new(DummyOutput) diff --git a/output_file.go b/output_file.go index ac0af4b..4f5be15 100644 --- a/output_file.go +++ b/output_file.go @@ -8,26 +8,30 @@ import ( "time" ) +// RawRequest stores original start time and request payload type RawRequest struct { Timestamp int64 Request []byte } +// FileOutput output plugin type FileOutput struct { path string encoder *gob.Encoder file *os.File } + +// NewFileOutput constructor for FileOutput, accepts path func NewFileOutput(path string) io.Writer { o := new(FileOutput) o.path = path - o.Init(path) + o.init(path) return o } -func (o *FileOutput) Init(path string) { +func (o *FileOutput) init(path string) { var err error o.file, err = os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660) diff --git a/plugins.go b/plugins.go index cb681ae..d47012b 100644 --- a/plugins.go +++ b/plugins.go @@ -13,7 +13,7 @@ type InOutPlugins struct { } // Plugins holds all the plugin objects -var Plugins *InOutPlugins +var Plugins *InOutPlugins = new(InOutPlugins) // extractLimitOptions detects if plugin get called with limiter support // Returns address and limit