mirror of
https://github.com/hanzoai/oxy.git
synced 2026-09-30 23:42:31 +00:00
Merge pull request #7 from mailgun/alexander/trace
Implement structured tracer
This commit is contained in:
Alexander Klizhentas
+1
-2
@@ -100,11 +100,10 @@ func MakeRequest(url string, opts ...ReqOption) (*http.Response, []byte, error)
|
||||
}
|
||||
}
|
||||
|
||||
method := "GET"
|
||||
if o.Method == "" {
|
||||
o.Method = "GET"
|
||||
}
|
||||
request, _ := http.NewRequest(method, url, strings.NewReader(o.Body))
|
||||
request, _ := http.NewRequest(o.Method, url, strings.NewReader(o.Body))
|
||||
if o.Headers != nil {
|
||||
utils.CopyHeaders(request.Header, o.Headers)
|
||||
}
|
||||
|
||||
+228
@@ -0,0 +1,228 @@
|
||||
// Package trace implement structured logging of requests
|
||||
package trace
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/mailgun/oxy/utils"
|
||||
)
|
||||
|
||||
// Option is a functional option setter for Tracer
|
||||
type Option func(*Tracer) error
|
||||
|
||||
// ErrorHandler is a functional argument that sets error handler of the server
|
||||
func ErrorHandler(h utils.ErrorHandler) Option {
|
||||
return func(t *Tracer) error {
|
||||
t.errHandler = h
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// RequestHeaders adds request headers to capture
|
||||
func RequestHeaders(headers ...string) Option {
|
||||
return func(t *Tracer) error {
|
||||
t.reqHeaders = append(t.reqHeaders, headers...)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// ResponseHeaders adds response headers to capture
|
||||
func ResponseHeaders(headers ...string) Option {
|
||||
return func(t *Tracer) error {
|
||||
t.respHeaders = append(t.respHeaders, headers...)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// Logger sets optional logger for trace used to report errors
|
||||
func Logger(l utils.Logger) Option {
|
||||
return func(t *Tracer) error {
|
||||
t.log = l
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// Tracer records request and response emitting JSON structured data to the output
|
||||
type Tracer struct {
|
||||
errHandler utils.ErrorHandler
|
||||
next http.Handler
|
||||
reqHeaders []string
|
||||
respHeaders []string
|
||||
writer io.Writer
|
||||
log utils.Logger
|
||||
}
|
||||
|
||||
// New creates a new Tracer middleware that emits all the request/response information in structured format
|
||||
// to writer and passes the request to the next handler. It can optionally capture request and response headers,
|
||||
// see RequestHeaders and ResponseHeaders options for details.
|
||||
func New(next http.Handler, writer io.Writer, opts ...Option) (*Tracer, error) {
|
||||
t := &Tracer{
|
||||
writer: writer,
|
||||
next: next,
|
||||
}
|
||||
for _, o := range opts {
|
||||
if err := o(t); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if t.errHandler == nil {
|
||||
t.errHandler = utils.DefaultHandler
|
||||
}
|
||||
if t.log == nil {
|
||||
t.log = utils.NullLogger
|
||||
}
|
||||
return t, nil
|
||||
}
|
||||
|
||||
func (t *Tracer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
||||
start := time.Now()
|
||||
pw := &utils.ProxyWriter{W: w}
|
||||
t.next.ServeHTTP(pw, req)
|
||||
|
||||
l := t.newRecord(req, pw, time.Since(start))
|
||||
if err := json.NewEncoder(t.writer).Encode(l); err != nil {
|
||||
t.log.Errorf("Failed to marshal request: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (t *Tracer) newRecord(req *http.Request, pw *utils.ProxyWriter, diff time.Duration) *Record {
|
||||
return &Record{
|
||||
Request: Request{
|
||||
Method: req.Method,
|
||||
URL: req.URL.String(),
|
||||
TLS: newTLS(req),
|
||||
BodyBytes: bodyBytes(req.Header),
|
||||
Headers: captureHeaders(req.Header, t.reqHeaders),
|
||||
},
|
||||
Response: Response{
|
||||
Code: pw.StatusCode(),
|
||||
BodyBytes: bodyBytes(pw.Header()),
|
||||
Roundtrip: float64(diff) / float64(time.Millisecond),
|
||||
Headers: captureHeaders(pw.Header(), t.respHeaders),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func newTLS(req *http.Request) *TLS {
|
||||
if req.TLS == nil {
|
||||
return nil
|
||||
}
|
||||
return &TLS{
|
||||
Version: versionToString(req.TLS.Version),
|
||||
Resume: req.TLS.DidResume,
|
||||
CipherSuite: csToString(req.TLS.CipherSuite),
|
||||
Server: req.TLS.ServerName,
|
||||
}
|
||||
}
|
||||
|
||||
func captureHeaders(in http.Header, headers []string) http.Header {
|
||||
if len(headers) == 0 || in == nil {
|
||||
return nil
|
||||
}
|
||||
out := make(http.Header, len(headers))
|
||||
for _, h := range headers {
|
||||
vals, ok := in[h]
|
||||
if !ok || len(out[h]) != 0 {
|
||||
continue
|
||||
}
|
||||
for i := range vals {
|
||||
out.Add(h, vals[i])
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Record represents a structured request and response record
|
||||
type Record struct {
|
||||
Request Request `json:"request"`
|
||||
Response Response `json:"response"`
|
||||
}
|
||||
|
||||
// Req contains information about an HTTP request
|
||||
type Request struct {
|
||||
Method string `json:"method"` // Method - request method
|
||||
BodyBytes int64 `json:"body_bytes"` // BodyBytes - size of request body in bytes
|
||||
URL string `json:"url"` // URL - Request URL
|
||||
Headers http.Header `json:"headers,omitempty"` // Headers - optional request headers, will be recorded if configured
|
||||
TLS *TLS `json:"tls,omitempty"` // TLS - optional TLS record, will be recorded if it's a TLS connection
|
||||
}
|
||||
|
||||
// Resp contains information about HTTP response
|
||||
type Response struct {
|
||||
Code int `json:"code"` // Code - response status code
|
||||
Roundtrip float64 `json:"roundtrip"` // Roundtrip - round trip time in milliseconds
|
||||
Headers http.Header `json:"headers,omitempty"` // Headers - optional headers, will be recorded if configured
|
||||
BodyBytes int64 `json:"body_bytes"` // BodyBytes - size of response body in bytes
|
||||
}
|
||||
|
||||
// TLS contains information about this TLS connection
|
||||
type TLS struct {
|
||||
Version string `json:"version"` // Version - TLS version
|
||||
Resume bool `json:"resume"` // Resume tells if the session has been re-used (session tickets)
|
||||
CipherSuite string `json:"cipher_suite"` // CipherSuite contains cipher suite used for this connection
|
||||
Server string `json:"server"` // Server contains server name used in SNI
|
||||
}
|
||||
|
||||
func versionToString(v uint16) string {
|
||||
switch v {
|
||||
case tls.VersionSSL30:
|
||||
return "SSL30"
|
||||
case tls.VersionTLS10:
|
||||
return "TLS10"
|
||||
case tls.VersionTLS11:
|
||||
return "TLS11"
|
||||
case tls.VersionTLS12:
|
||||
return "TLS12"
|
||||
}
|
||||
return fmt.Sprintf("unknown: %x", v)
|
||||
}
|
||||
|
||||
func csToString(cs uint16) string {
|
||||
switch cs {
|
||||
case tls.TLS_RSA_WITH_RC4_128_SHA:
|
||||
return "TLS_RSA_WITH_RC4_128_SHA"
|
||||
case tls.TLS_RSA_WITH_3DES_EDE_CBC_SHA:
|
||||
return "TLS_RSA_WITH_3DES_EDE_CBC_SHA"
|
||||
case tls.TLS_RSA_WITH_AES_128_CBC_SHA:
|
||||
return "TLS_RSA_WITH_AES_128_CBC_SHA"
|
||||
case tls.TLS_RSA_WITH_AES_256_CBC_SHA:
|
||||
return "TLS_RSA_WITH_AES_256_CBC_SHA"
|
||||
case tls.TLS_ECDHE_ECDSA_WITH_RC4_128_SHA:
|
||||
return "TLS_ECDHE_ECDSA_WITH_RC4_128_SHA"
|
||||
case tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA:
|
||||
return "TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA"
|
||||
case tls.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA:
|
||||
return "TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA"
|
||||
case tls.TLS_ECDHE_RSA_WITH_RC4_128_SHA:
|
||||
return "TLS_ECDHE_RSA_WITH_RC4_128_SHA"
|
||||
case tls.TLS_ECDHE_RSA_WITH_3DES_EDE_CBC_SHA:
|
||||
return "TLS_ECDHE_RSA_WITH_3DES_EDE_CBC_SHA"
|
||||
case tls.TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA:
|
||||
return "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA"
|
||||
case tls.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA:
|
||||
return "TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA"
|
||||
case tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256:
|
||||
return "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256"
|
||||
case tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256:
|
||||
return "TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256"
|
||||
}
|
||||
return fmt.Sprintf("unknown: %x", cs)
|
||||
}
|
||||
|
||||
func bodyBytes(h http.Header) int64 {
|
||||
len := h.Get("Content-Length")
|
||||
if len == "" {
|
||||
return 0
|
||||
}
|
||||
bytes, err := strconv.ParseInt(len, 10, 0)
|
||||
if err == nil {
|
||||
return bytes
|
||||
}
|
||||
return 0
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
package trace
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"crypto/tls"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"testing"
|
||||
|
||||
"github.com/mailgun/oxy/testutils"
|
||||
"github.com/mailgun/oxy/utils"
|
||||
|
||||
. "gopkg.in/check.v1"
|
||||
)
|
||||
|
||||
func TestTrace(t *testing.T) { TestingT(t) }
|
||||
|
||||
type TraceSuite struct{}
|
||||
|
||||
var _ = Suite(&TraceSuite{})
|
||||
|
||||
func (s *TraceSuite) TestTraceSimple(c *C) {
|
||||
handler := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
||||
w.Header().Set("Content-Length", "5")
|
||||
w.Write([]byte("hello"))
|
||||
})
|
||||
buf := &bytes.Buffer{}
|
||||
l := utils.NewFileLogger(buf, utils.INFO)
|
||||
|
||||
trace := &bytes.Buffer{}
|
||||
|
||||
t, err := New(handler, trace, Logger(l))
|
||||
c.Assert(err, IsNil)
|
||||
|
||||
srv := httptest.NewServer(t)
|
||||
defer srv.Close()
|
||||
|
||||
re, _, err := testutils.MakeRequest(srv.URL+"/hello", testutils.Method("POST"), testutils.Body("123456"))
|
||||
c.Assert(err, IsNil)
|
||||
c.Assert(re.StatusCode, Equals, http.StatusOK)
|
||||
|
||||
var r *Record
|
||||
c.Assert(json.Unmarshal(trace.Bytes(), &r), IsNil)
|
||||
|
||||
c.Assert(r.Request.Method, Equals, "POST")
|
||||
c.Assert(r.Request.URL, Equals, "/hello")
|
||||
c.Assert(r.Response.Code, Equals, http.StatusOK)
|
||||
c.Assert(r.Request.BodyBytes, Equals, int64(6))
|
||||
c.Assert(r.Response.Roundtrip, Not(Equals), float64(0))
|
||||
c.Assert(r.Response.BodyBytes, Equals, int64(5))
|
||||
}
|
||||
|
||||
func (s *TraceSuite) TestTraceCaptureHeaders(c *C) {
|
||||
respHeaders := http.Header{
|
||||
"X-Re-1": []string{"6", "7"},
|
||||
"X-Re-2": []string{"2", "3"},
|
||||
}
|
||||
|
||||
handler := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
||||
utils.CopyHeaders(w.Header(), respHeaders)
|
||||
w.Write([]byte("hello"))
|
||||
})
|
||||
buf := &bytes.Buffer{}
|
||||
l := utils.NewFileLogger(buf, utils.INFO)
|
||||
|
||||
trace := &bytes.Buffer{}
|
||||
|
||||
t, err := New(handler, trace, Logger(l), RequestHeaders("X-Req-B", "X-Req-A"), ResponseHeaders("X-Re-1", "X-Re-2"))
|
||||
c.Assert(err, IsNil)
|
||||
|
||||
srv := httptest.NewServer(t)
|
||||
defer srv.Close()
|
||||
|
||||
reqHeaders := http.Header{"X-Req-A": []string{"1", "2"}, "X-Req-B": []string{"3", "4"}}
|
||||
re, _, err := testutils.Get(srv.URL+"/hello", testutils.Headers(reqHeaders))
|
||||
c.Assert(err, IsNil)
|
||||
c.Assert(re.StatusCode, Equals, http.StatusOK)
|
||||
|
||||
var r *Record
|
||||
c.Assert(json.Unmarshal(trace.Bytes(), &r), IsNil)
|
||||
|
||||
c.Assert(r.Request.Headers, DeepEquals, reqHeaders)
|
||||
c.Assert(r.Response.Headers, DeepEquals, respHeaders)
|
||||
}
|
||||
|
||||
func (s *TraceSuite) TestTraceTLS(c *C) {
|
||||
handler := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
||||
w.Write([]byte("hello"))
|
||||
})
|
||||
buf := &bytes.Buffer{}
|
||||
l := utils.NewFileLogger(buf, utils.INFO)
|
||||
|
||||
trace := &bytes.Buffer{}
|
||||
|
||||
t, err := New(handler, trace, Logger(l))
|
||||
c.Assert(err, IsNil)
|
||||
|
||||
srv := httptest.NewUnstartedServer(t)
|
||||
srv.StartTLS()
|
||||
defer srv.Close()
|
||||
|
||||
config := &tls.Config{
|
||||
InsecureSkipVerify: true,
|
||||
}
|
||||
|
||||
u, err := url.Parse(srv.URL)
|
||||
c.Assert(err, IsNil)
|
||||
|
||||
conn, err := tls.Dial("tcp", u.Host, config)
|
||||
c.Assert(err, IsNil)
|
||||
|
||||
fmt.Fprintf(conn, "GET / HTTP/1.0\r\n\r\n")
|
||||
status, err := bufio.NewReader(conn).ReadString('\n')
|
||||
c.Assert(status, Equals, "HTTP/1.0 200 OK\r\n")
|
||||
state := conn.ConnectionState()
|
||||
conn.Close()
|
||||
|
||||
var r *Record
|
||||
c.Assert(json.Unmarshal(trace.Bytes(), &r), IsNil)
|
||||
c.Assert(r.Request.TLS.Version, Equals, versionToString(state.Version))
|
||||
}
|
||||
@@ -6,11 +6,23 @@ import (
|
||||
"net/url"
|
||||
)
|
||||
|
||||
// ProxyWriter helps to capture response headers and status code
|
||||
// from the ServeHTTP. It can be safely passed to ServeHTTP handler,
|
||||
// wrapping the real response writer.
|
||||
type ProxyWriter struct {
|
||||
W http.ResponseWriter
|
||||
Code int
|
||||
}
|
||||
|
||||
func (p *ProxyWriter) StatusCode() int {
|
||||
if p.Code == 0 {
|
||||
// per contract standard lib will set this to http.StatusOK if not set
|
||||
// by user, here we avoid the confusion by mirroring this logic
|
||||
return http.StatusOK
|
||||
}
|
||||
return p.Code
|
||||
}
|
||||
|
||||
func (p *ProxyWriter) Header() http.Header {
|
||||
return p.W.Header()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user