hanzoai/oxy
2
0
Fork 0
mirror of https://github.com/hanzoai/oxy.git synced 2026-09-30 23:42:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Drop internal log lib, use logrus (#50)

Browse Source 
* Drop internal log lib, use logrus

* Removed merge artifacts

* Bad merge

* More missing ctx.log references
This commit is contained in:
Archis
2016-08-30 08:24:20 -07:00
committed by GitHub
parent 4bc0c209b8
commit 34e2ebf3d7
17 changed files with 59 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cbreaker/cbreaker.go
+5 -14
View File
@@ -31,6 +31,7 @@ import (
"sync"
"time"
log "github.com/Sirupsen/logrus"
"github.com/mailgun/timetools"
"github.com/vulcand/oxy/memmetrics"
"github.com/vulcand/oxy/utils"
@@ -60,7 +61,6 @@ type CircuitBreaker struct {
fallback http.Handler
next http.Handler
log utils.Logger
clock timetools.TimeProvider
}
@@ -75,7 +75,6 @@ func New(next http.Handler, expression string, options ...CircuitBreakerOption)
fallbackDuration: defaultFallbackDuration,
recoveryDuration: defaultRecoveryDuration,
fallback: defaultFallback,
log: utils.NullLogger,
}
for _, s := range options {
@@ -121,7 +120,7 @@ func (c *CircuitBreaker) activateFallback(w http.ResponseWriter, req *http.Reque
c.m.Lock()
defer c.m.Unlock()
c.log.Infof("%v is in error state", c)
log.Infof("%v is in error state", c)
switch c.state {
case stateStandby:
@@ -186,13 +185,13 @@ func (c *CircuitBreaker) exec(s SideEffect) {
}
go func() {
if err := s.Exec(); err != nil {
c.log.Errorf("%v side effect failure: %v", c, err)
log.Errorf("%v side effect failure: %v", c, err)
}
}()
}
func (c *CircuitBreaker) setState(new cbState, until time.Time) {
c.log.Infof("%v setting state to %v, until %v", c, new, until)
log.Infof("%v setting state to %v, until %v", c, new, until)
c.state = new
c.until = until
switch new {
@@ -225,7 +224,7 @@ func (c *CircuitBreaker) checkAndSet() {
c.lastCheck = c.clock.UtcNow().Add(c.checkPeriod)
if c.state == stateTripped {
c.log.Infof("%v skip set tripped", c)
log.Infof("%v skip set tripped", c)
return
}
@@ -309,14 +308,6 @@ func Fallback(h http.Handler) CircuitBreakerOption {
}
}
// Logger adds logging for the CircuitBreaker.
func Logger(l utils.Logger) CircuitBreakerOption {
return func(c *CircuitBreaker) error {
c.log = l
return nil
}
}
// cbState is the state of the circuit breaker
type cbState int
cbreaker/predicates.go
+2 -1
View File
@@ -4,6 +4,7 @@ import (
"fmt"
"time"
log "github.com/Sirupsen/logrus"
"github.com/vulcand/predicate"
)
@@ -49,7 +50,7 @@ func latencyAtQuantile(quantile float64) toInt {
return func(c *CircuitBreaker) int {
h, err := c.metrics.LatencyHistogram()
if err != nil {
c.log.Errorf("Failed to get latency histogram, for %v error: %v", c, err)
log.Errorf("Failed to get latency histogram, for %v error: %v", c, err)
return 0
}
return int(h.LatencyAtQuantile(quantile) / time.Millisecond)
connlimit/connlimit.go
+3 -14
View File
@@ -6,6 +6,7 @@ import (
"net/http"
"sync"
log "github.com/Sirupsen/logrus"
"github.com/vulcand/oxy/utils"
)
@@ -20,7 +21,6 @@ type ConnLimiter struct {
next http.Handler
errHandler utils.ErrorHandler
log utils.Logger
}
func New(next http.Handler, extract utils.SourceExtractor, maxConnections int64, options ...ConnLimitOption) (*ConnLimiter, error) {
@@ -40,9 +40,6 @@ func New(next http.Handler, extract utils.SourceExtractor, maxConnections int64,
return nil, err
}
}
if cl.log == nil {
cl.log = utils.NullLogger
}
if cl.errHandler == nil {
cl.errHandler = defaultErrHandler
}
@@ -56,12 +53,12 @@ func (cl *ConnLimiter) Wrap(h http.Handler) {
func (cl *ConnLimiter) ServeHTTP(w http.ResponseWriter, r *http.Request) {
token, amount, err := cl.extract.Extract(r)
if err != nil {
cl.log.Errorf("failed to extract source of the connection: %v", err)
log.Errorf("failed to extract source of the connection: %v", err)
cl.errHandler.ServeHTTP(w, r, err)
return
}
if err := cl.acquire(token, amount); err != nil {
cl.log.Infof("limiting request source %s: %v", token, err)
log.Infof("limiting request source %s: %v", token, err)
cl.errHandler.ServeHTTP(w, r, err)
return
}
@@ -120,14 +117,6 @@ func (e *ConnErrHandler) ServeHTTP(w http.ResponseWriter, req *http.Request, err
type ConnLimitOption func(l *ConnLimiter) error
// Logger sets the logger that will be used by this middleware.
func Logger(l utils.Logger) ConnLimitOption {
return func(cl *ConnLimiter) error {
cl.log = l
return nil
}
}
// ErrorHandler sets error handler of the server
func ErrorHandler(h utils.ErrorHandler) ConnLimitOption {
return func(cl *ConnLimiter) error {
connlimit/connlimit_test.go
+1 -10
View File
@@ -1,7 +1,6 @@
package connlimit
import (
"bytes"
"fmt"
"net/http"
"net/http/httptest"
@@ -20,9 +19,6 @@ type ConnLimiterSuite struct {
var _ = Suite(&ConnLimiterSuite{})
func (s *ConnLimiterSuite) SetUpSuite(c *C) {
}
// We've hit the limit and were able to proceed once the request has completed
func (s *ConnLimiterSuite) TestHitLimitAndRelease(c *C) {
wait := make(chan bool)
@@ -77,10 +73,7 @@ func (s *ConnLimiterSuite) TestCustomHandlers(c *C) {
w.Write([]byte(http.StatusText(http.StatusTeapot)))
})
buf := &bytes.Buffer{}
log := utils.NewFileLogger(buf, utils.INFO)
l, err := New(handler, headerLimit, 0, ErrorHandler(errHandler), Logger(log))
l, err := New(handler, headerLimit, 0, ErrorHandler(errHandler))
c.Assert(err, Equals, nil)
srv := httptest.NewServer(l)
@@ -89,8 +82,6 @@ func (s *ConnLimiterSuite) TestCustomHandlers(c *C) {
re, _, err := testutils.Get(srv.URL, testutils.Header("Limit", "a"))
c.Assert(err, IsNil)
c.Assert(re.StatusCode, Equals, http.StatusTeapot)
c.Assert(len(buf.String()), Not(Equals), 0)
}
// We've hit the limit and were able to proceed once the request has completed
forward/fwd.go
+9 -23
View File
@@ -14,6 +14,7 @@ import (
"strings"
"time"
log "github.com/Sirupsen/logrus"
"github.com/vulcand/oxy/utils"
)
@@ -79,17 +80,6 @@ func ErrorHandler(h utils.ErrorHandler) optSetter {
}
}
// Logger specifies the logger to use.
// Forwarder will default to oxyutils.NullLogger if no logger has been specified
func Logger(l utils.Logger) optSetter {
return func(f *Forwarder) error {
f.log = l
return nil
}
}
// Forwarder wraps two traffic forwarding implementations: HTTP and websockets.
// It decides based on the specified request which implementation to use
type Forwarder struct {
*httpForwarder
*websocketForwarder
@@ -99,7 +89,6 @@ type Forwarder struct {
// handlerContext defines a handler context for error reporting and logging
type handlerContext struct {
errHandler utils.ErrorHandler
log utils.Logger
}
// httpForwarder is a handler that can reverse proxy
@@ -143,9 +132,6 @@ func New(setters ...optSetter) (*Forwarder, error) {
}
f.httpForwarder.rewriter = &HeaderRewriter{TrustForwardHeader: true, Hostname: h}
}
if f.log == nil {
f.log = utils.NullLogger
}
if f.errHandler == nil {
f.errHandler = utils.DefaultHandler
}
@@ -167,20 +153,20 @@ func (f *httpForwarder) serveHTTP(w http.ResponseWriter, req *http.Request, ctx
start := time.Now().UTC()
response, err := f.roundTripper.RoundTrip(f.copyRequest(req, req.URL))
if err != nil {
ctx.log.Errorf("Error forwarding to %v, err: %v", req.URL, err)
log.Errorf("Error forwarding to %v, err: %v", req.URL, err)
ctx.errHandler.ServeHTTP(w, req, err)
return
}
if req.TLS != nil {
ctx.log.Infof("Round trip: %v, code: %v, duration: %v tls:version: %x, tls:resume:%t, tls:csuite:%x, tls:server:%v",
log.Infof("Round trip: %v, code: %v, duration: %v tls:version: %x, tls:resume:%t, tls:csuite:%x, tls:server:%v",
req.URL, response.StatusCode, time.Now().UTC().Sub(start),
req.TLS.Version,
req.TLS.DidResume,
req.TLS.CipherSuite,
req.TLS.ServerName)
} else {
ctx.log.Infof("Round trip: %v, code: %v, duration: %v",
log.Infof("Round trip: %v, code: %v, duration: %v",
req.URL, response.StatusCode, time.Now().UTC().Sub(start))
}
@@ -190,7 +176,7 @@ func (f *httpForwarder) serveHTTP(w http.ResponseWriter, req *http.Request, ctx
defer response.Body.Close()
if err != nil {
ctx.log.Errorf("Error copying upstream response Body: %v", err)
log.Errorf("Error copying upstream response Body: %v", err)
ctx.errHandler.ServeHTTP(w, req, err)
return
}
@@ -248,19 +234,19 @@ func (f *websocketForwarder) serveHTTP(w http.ResponseWriter, req *http.Request,
targetConn, err := f.dial("tcp", host)
if err != nil {
ctx.log.Errorf("Error dialing `%v`: %v", host, err)
log.Errorf("Error dialing `%v`: %v", host, err)
ctx.errHandler.ServeHTTP(w, req, err)
return
}
hijacker, ok := w.(http.Hijacker)
if !ok {
ctx.log.Errorf("Unable to hijack the connection: does not implement http.Hijacker")
log.Errorf("Unable to hijack the connection: does not implement http.Hijacker")
ctx.errHandler.ServeHTTP(w, req, err)
return
}
underlyingConn, _, err := hijacker.Hijack()
if err != nil {
ctx.log.Errorf("Unable to hijack the connection: %v", err)
log.Errorf("Unable to hijack the connection: %v", err)
ctx.errHandler.ServeHTTP(w, req, err)
return
}
@@ -270,7 +256,7 @@ func (f *websocketForwarder) serveHTTP(w http.ResponseWriter, req *http.Request,
// write the modified incoming request to the dialed connection
if err = outReq.Write(targetConn); err != nil {
ctx.log.Errorf("Unable to copy request to target: %v", err)
log.Errorf("Unable to copy request to target: %v", err)
ctx.errHandler.ServeHTTP(w, req, err)
return
}
forward/fwd_test.go
+2 -12
View File
@@ -1,7 +1,6 @@
package forward
import (
"bytes"
"fmt"
"net"
"net/http"
@@ -188,10 +187,7 @@ func (s *FwdSuite) TestCustomLogger(c *C) {
})
defer srv.Close()
buf := &bytes.Buffer{}
l := utils.NewFileLogger(buf, utils.INFO)
f, err := New(Logger(l))
f, err := New()
c.Assert(err, IsNil)
proxy := testutils.NewHandler(func(w http.ResponseWriter, req *http.Request) {
@@ -203,7 +199,6 @@ func (s *FwdSuite) TestCustomLogger(c *C) {
re, _, err := testutils.Get(proxy.URL)
c.Assert(err, IsNil)
c.Assert(re.StatusCode, Equals, http.StatusOK)
c.Assert(strings.Contains(buf.String(), srv.URL), Equals, true)
}
func (s *FwdSuite) TestEscapedURL(c *C) {
@@ -243,10 +238,7 @@ func (s *FwdSuite) TestForwardedProto(c *C) {
})
defer srv.Close()
buf := &bytes.Buffer{}
l := utils.NewFileLogger(buf, utils.INFO)
f, err := New(Logger(l))
f, err := New()
c.Assert(err, IsNil)
proxy := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
@@ -261,8 +253,6 @@ func (s *FwdSuite) TestForwardedProto(c *C) {
c.Assert(err, IsNil)
c.Assert(re.StatusCode, Equals, http.StatusOK)
c.Assert(proto, Equals, "https")
c.Assert(strings.Contains(buf.String(), "tls"), Equals, true)
}
func (s *FwdSuite) TestChunkedResponseConversion(c *C) {
ratelimit/tokenlimiter.go
+3 -14
View File
@@ -7,6 +7,7 @@ import (
"sync"
"time"
log "github.com/Sirupsen/logrus"
"github.com/mailgun/timetools"
"github.com/mailgun/ttlmap"
"github.com/vulcand/oxy/utils"
@@ -65,7 +66,6 @@ type TokenLimiter struct {
mutex sync.Mutex
bucketSets *ttlmap.TtlMap
errHandler utils.ErrorHandler
log utils.Logger
capacity int
next http.Handler
}
@@ -110,7 +110,7 @@ func (tl *TokenLimiter) ServeHTTP(w http.ResponseWriter, req *http.Request) {
}
if err := tl.consumeRates(req, source, amount); err != nil {
tl.log.Infof("limiting request %v %v, limit: %v", req.Method, req.URL, err)
log.Infof("limiting request %v %v, limit: %v", req.Method, req.URL, err)
tl.errHandler.ServeHTTP(w, req, err)
return
}
@@ -155,7 +155,7 @@ func (tl *TokenLimiter) resolveRates(req *http.Request) *RateSet {
rates, err := tl.extractRates.Extract(req)
if err != nil {
tl.log.Errorf("Failed to retrieve rates: %v", err)
log.Errorf("Failed to retrieve rates: %v", err)
return tl.defaultRates
}
@@ -190,14 +190,6 @@ func (e *RateErrHandler) ServeHTTP(w http.ResponseWriter, req *http.Request, err
type TokenLimiterOption func(l *TokenLimiter) error
// Logger sets the logger that will be used by this middleware.
func Logger(l utils.Logger) TokenLimiterOption {
return func(cl *TokenLimiter) error {
cl.log = l
return nil
}
}
// ErrorHandler sets error handler of the server
func ErrorHandler(h utils.ErrorHandler) TokenLimiterOption {
return func(cl *TokenLimiter) error {
@@ -233,9 +225,6 @@ func Capacity(cap int) TokenLimiterOption {
var defaultErrHandler = &RateErrHandler{}
func setDefaults(tl *TokenLimiter) {
if tl.log == nil {
tl.log = utils.NullLogger
}
if tl.capacity <= 0 {
tl.capacity = DefaultCapacity
}
ratelimit/tokenlimiter_test.go
+1 -7
View File
@@ -1,7 +1,6 @@
package ratelimit
import (
"bytes"
"fmt"
"net/http"
"net/http/httptest"
@@ -299,10 +298,7 @@ func (s *LimiterSuite) TestOptions(c *C) {
w.Write([]byte(http.StatusText(http.StatusTeapot)))
})
buf := &bytes.Buffer{}
log := utils.NewFileLogger(buf, utils.INFO)
l, err := New(handler, headerLimit, rates, ErrorHandler(errHandler), Logger(log), Clock(s.clock))
l, err := New(handler, headerLimit, rates, ErrorHandler(errHandler), Clock(s.clock))
c.Assert(err, IsNil)
srv := httptest.NewServer(l)
@@ -315,8 +311,6 @@ func (s *LimiterSuite) TestOptions(c *C) {
re, _, err = testutils.Get(srv.URL, testutils.Header("Source", "a"))
c.Assert(err, IsNil)
c.Assert(re.StatusCode, Equals, http.StatusTeapot)
c.Assert(len(buf.String()), Not(Equals), 0)
}
func headerLimiter(req *http.Request) (string, int64, error) {
roundrobin/rebalancer.go
+5 -16
View File
@@ -7,6 +7,7 @@ import (
"sync"
"time"
log "github.com/Sirupsen/logrus"
"github.com/mailgun/timetools"
"github.com/vulcand/oxy/memmetrics"
"github.com/vulcand/oxy/utils"
@@ -42,21 +43,12 @@ type Rebalancer struct {
// errHandler is HTTP handler called in case of errors
errHandler utils.ErrorHandler
log utils.Logger
ratings []float64
// creates new meters
newMeter NewMeterFn
}
func RebalancerLogger(log utils.Logger) RebalancerOption {
return func(r *Rebalancer) error {
r.log = log
return nil
}
}
func RebalancerClock(clock timetools.TimeProvider) RebalancerOption {
return func(r *Rebalancer) error {
r.clock = clock
@@ -102,9 +94,6 @@ func NewRebalancer(handler balancerHandler, opts ...RebalancerOption) (*Rebalanc
if rb.backoffDuration == 0 {
rb.backoffDuration = 10 * time.Second
}
if rb.log == nil {
rb.log = &utils.NOPLogger{}
}
if rb.newMeter == nil {
rb.newMeter = func() (Meter, error) {
rc, err := memmetrics.NewRatioCounter(10, time.Second, memmetrics.RatioClock(rb.clock))
@@ -270,7 +259,7 @@ func (rb *Rebalancer) adjustWeights() {
func (rb *Rebalancer) applyWeights() {
for _, srv := range rb.servers {
rb.log.Infof("upsert server %v, weight %v", srv.url, srv.curWeight)
log.Infof("upsert server %v, weight %v", srv.url, srv.curWeight)
rb.next.UpsertServer(srv.url, Weight(srv.curWeight))
}
}
@@ -282,7 +271,7 @@ func (rb *Rebalancer) setMarkedWeights() bool {
if srv.good {
weight := increase(srv.curWeight)
if weight <= FSMMaxWeight {
rb.log.Infof("increasing weight of %v from %v to %v", srv.url, srv.curWeight, weight)
log.Infof("increasing weight of %v from %v to %v", srv.url, srv.curWeight, weight)
srv.curWeight = weight
changed = true
}
@@ -329,7 +318,7 @@ func (rb *Rebalancer) markServers() bool {
}
}
if len(g) != 0 && len(b) != 0 {
rb.log.Infof("bad: %v good: %v, ratings: %v", b, g, rb.ratings)
log.Infof("bad: %v good: %v, ratings: %v", b, g, rb.ratings)
}
return len(g) != 0 && len(b) != 0
}
@@ -343,7 +332,7 @@ func (rb *Rebalancer) convergeWeights() bool {
}
changed = true
newWeight := decrease(s.origWeight, s.curWeight)
rb.log.Infof("decreasing weight of %v from %v to %v", s.url, s.curWeight, newWeight)
log.Infof("decreasing weight of %v from %v to %v", s.url, s.curWeight, newWeight)
s.curWeight = newWeight
}
if !changed {
roundrobin/rebalancer_test.go
+4 -11
View File
@@ -3,20 +3,17 @@ package roundrobin
import (
"net/http"
"net/http/httptest"
"os"
"time"
"github.com/mailgun/timetools"
"github.com/vulcand/oxy/forward"
"github.com/vulcand/oxy/testutils"
"github.com/vulcand/oxy/utils"
. "gopkg.in/check.v1"
)
type RBSuite struct {
clock *timetools.FreezedTime
log utils.Logger
}
var _ = Suite(&RBSuite{
@@ -25,10 +22,6 @@ var _ = Suite(&RBSuite{
},
})
func (s *RBSuite) SetUpSuite(c *C) {
s.log = utils.NewFileLogger(os.Stdout, utils.INFO)
}
func (s *RBSuite) TestRebalancerNormalOperation(c *C) {
a, b := testutils.NewResponder("a"), testutils.NewResponder("b")
defer a.Close()
@@ -111,7 +104,7 @@ func (s *RBSuite) TestRebalancerRecovery(c *C) {
newMeter := func() (Meter, error) {
return &testMeter{}, nil
}
rb, err := NewRebalancer(lb, RebalancerMeter(newMeter), RebalancerClock(s.clock), RebalancerLogger(s.log))
rb, err := NewRebalancer(lb, RebalancerMeter(newMeter), RebalancerClock(s.clock))
c.Assert(err, IsNil)
rb.UpsertServer(testutils.ParseURI(a.URL))
@@ -166,7 +159,7 @@ func (s *RBSuite) TestRebalancerCascading(c *C) {
newMeter := func() (Meter, error) {
return &testMeter{}, nil
}
rb, err := NewRebalancer(lb, RebalancerMeter(newMeter), RebalancerClock(s.clock), RebalancerLogger(s.log))
rb, err := NewRebalancer(lb, RebalancerMeter(newMeter), RebalancerClock(s.clock))
c.Assert(err, IsNil)
rb.UpsertServer(testutils.ParseURI(a.URL))
@@ -221,7 +214,7 @@ func (s *RBSuite) TestRebalancerAllBad(c *C) {
newMeter := func() (Meter, error) {
return &testMeter{}, nil
}
rb, err := NewRebalancer(lb, RebalancerMeter(newMeter), RebalancerClock(s.clock), RebalancerLogger(s.log))
rb, err := NewRebalancer(lb, RebalancerMeter(newMeter), RebalancerClock(s.clock))
c.Assert(err, IsNil)
rb.UpsertServer(testutils.ParseURI(a.URL))
@@ -262,7 +255,7 @@ func (s *RBSuite) TestRebalancerReset(c *C) {
newMeter := func() (Meter, error) {
return &testMeter{}, nil
}
rb, err := NewRebalancer(lb, RebalancerMeter(newMeter), RebalancerClock(s.clock), RebalancerLogger(s.log))
rb, err := NewRebalancer(lb, RebalancerMeter(newMeter), RebalancerClock(s.clock))
c.Assert(err, IsNil)
rb.UpsertServer(testutils.ParseURI(a.URL))
stream/retry_test.go
+2 -5
View File
@@ -3,12 +3,10 @@ package stream
import (
"net/http"
"net/http/httptest"
"os"
"github.com/vulcand/oxy/forward"
"github.com/vulcand/oxy/roundrobin"
"github.com/vulcand/oxy/testutils"
"github.com/vulcand/oxy/utils"
. "gopkg.in/check.v1"
)
@@ -78,9 +76,8 @@ func (s *RTSuite) TestRetryExceedAttempts(c *C) {
}
func new(c *C, p string) (*roundrobin.RoundRobin, *Streamer) {
logger := utils.NewFileLogger(os.Stdout, utils.INFO)
// forwarder will proxy the request to whatever destination
fwd, err := forward.New(forward.Logger(logger))
fwd, err := forward.New()
c.Assert(err, IsNil)
// load balancer will round robin request
@@ -88,7 +85,7 @@ func new(c *C, p string) (*roundrobin.RoundRobin, *Streamer) {
c.Assert(err, IsNil)
// stream handler will forward requests to redirect, make sure it uses files
st, err := New(lb, Logger(logger), Retry(p), MemRequestBodyBytes(1))
st, err := New(lb, Retry(p), MemRequestBodyBytes(1))
c.Assert(err, IsNil)
return lb, st
stream/stream.go
+7 -19
View File
@@ -41,6 +41,7 @@ import (
"io/ioutil"
"net/http"
log "github.com/Sirupsen/logrus"
"github.com/mailgun/multibuf"
"github.com/vulcand/oxy/utils"
)
@@ -69,7 +70,6 @@ type Streamer struct {
next http.Handler
errHandler utils.ErrorHandler
log utils.Logger
}
// New returns a new streamer middleware. New() function supports optional functional arguments
@@ -92,10 +92,6 @@ func New(next http.Handler, setters ...optSetter) (*Streamer, error) {
strm.errHandler = errHandler
}
if strm.log == nil {
strm.log = utils.NullLogger
}
return strm, nil
}
@@ -123,14 +119,6 @@ func Retry(predicate string) optSetter {
}
}
// Logger sets the logger that will be used by this middleware.
func Logger(l utils.Logger) optSetter {
return func(s *Streamer) error {
s.log = l
return nil
}
}
// ErrorHandler sets error handler of the server
func ErrorHandler(h utils.ErrorHandler) optSetter {
return func(s *Streamer) error {
@@ -193,7 +181,7 @@ func (s *Streamer) Wrap(next http.Handler) error {
func (s *Streamer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
if err := s.checkLimit(req); err != nil {
s.log.Infof("request body over limit: %v", err)
log.Infof("request body over limit: %v", err)
s.errHandler.ServeHTTP(w, req, err)
return
}
@@ -217,7 +205,7 @@ func (s *Streamer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
// set without content length or using chunked TransferEncoding
totalSize, err := body.Size()
if err != nil {
s.log.Errorf("failed to get size, err %v", err)
log.Errorf("failed to get size, err %v", err)
s.errHandler.ServeHTTP(w, req, err)
return
}
@@ -246,7 +234,7 @@ func (s *Streamer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
if b.expectBody(outreq) {
rdr, err := writer.Reader()
if err != nil {
s.log.Errorf("failed to read response, err %v", err)
log.Errorf("failed to read response, err %v", err)
s.errHandler.ServeHTTP(w, req, err)
return
}
@@ -255,7 +243,7 @@ func (s *Streamer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
}
if (s.retryPredicate == nil || attempt > DefaultMaxRetryAttempts) ||
!s.retryPredicate(&context{r: req, attempt: attempt, responseCode: b.code, log: s.log}) {
!s.retryPredicate(&context{r: req, attempt: attempt, responseCode: b.code}) {
utils.CopyHeaders(w.Header(), b.Header())
w.WriteHeader(b.code)
if reader != nil {
@@ -266,12 +254,12 @@ func (s *Streamer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
attempt += 1
if _, err := body.Seek(0, 0); err != nil {
s.log.Errorf("Failed to rewind: error: %v", err)
log.Errorf("Failed to rewind: error: %v", err)
s.errHandler.ServeHTTP(w, req, err)
return
}
outreq = s.copyRequest(req, body, totalSize)
s.log.Infof("retry Request(%v %v) attempt %v", req.Method, req.URL, attempt)
log.Infof("retry Request(%v %v) attempt %v", req.Method, req.URL, attempt)
}
}
stream/stream_test.go
+10 -11
View File
@@ -8,7 +8,6 @@ import (
"net"
"net/http"
"net/http/httptest"
"os"
"testing"
"github.com/vulcand/oxy/forward"
@@ -41,7 +40,7 @@ func (s *STSuite) TestSimple(c *C) {
})
// stream handler will forward requests to redirect
st, err := New(rdr, Logger(utils.NewFileLogger(os.Stdout, utils.INFO)))
st, err := New(rdr)
c.Assert(err, IsNil)
proxy := httptest.NewServer(st)
@@ -76,7 +75,7 @@ func (s *STSuite) TestChunkedEncodingSuccess(c *C) {
})
// stream handler will forward requests to redirect
st, err := New(rdr, Logger(utils.NewFileLogger(os.Stdout, utils.INFO)))
st, err := New(rdr)
c.Assert(err, IsNil)
proxy := httptest.NewServer(st)
@@ -109,7 +108,7 @@ func (s *STSuite) TestChunkedEncodingLimitReached(c *C) {
})
// stream handler will forward requests to redirect
st, err := New(rdr, Logger(utils.NewFileLogger(os.Stdout, utils.INFO)), MemRequestBodyBytes(4), MaxRequestBodyBytes(8))
st, err := New(rdr, MemRequestBodyBytes(4), MaxRequestBodyBytes(8))
c.Assert(err, IsNil)
proxy := httptest.NewServer(st)
@@ -140,7 +139,7 @@ func (s *STSuite) TestRequestLimitReached(c *C) {
})
// stream handler will forward requests to redirect
st, err := New(rdr, Logger(utils.NewFileLogger(os.Stdout, utils.INFO)), MaxRequestBodyBytes(4))
st, err := New(rdr, MaxRequestBodyBytes(4))
c.Assert(err, IsNil)
proxy := httptest.NewServer(st)
@@ -168,7 +167,7 @@ func (s *STSuite) TestResponseLimitReached(c *C) {
})
// stream handler will forward requests to redirect
st, err := New(rdr, Logger(utils.NewFileLogger(os.Stdout, utils.INFO)), MaxResponseBodyBytes(4))
st, err := New(rdr, MaxResponseBodyBytes(4))
c.Assert(err, IsNil)
proxy := httptest.NewServer(st)
@@ -196,7 +195,7 @@ func (s *STSuite) TestFileStreamingResponse(c *C) {
})
// stream handler will forward requests to redirect
st, err := New(rdr, Logger(utils.NewFileLogger(os.Stdout, utils.INFO)), MemResponseBodyBytes(4))
st, err := New(rdr, MemResponseBodyBytes(4))
c.Assert(err, IsNil)
proxy := httptest.NewServer(st)
@@ -229,7 +228,7 @@ func (s *STSuite) TestCustomErrorHandler(c *C) {
w.WriteHeader(http.StatusTeapot)
w.Write([]byte(http.StatusText(http.StatusTeapot)))
})
st, err := New(rdr, Logger(utils.NewFileLogger(os.Stdout, utils.INFO)), MaxResponseBodyBytes(4), ErrorHandler(errHandler))
st, err := New(rdr, MaxResponseBodyBytes(4), ErrorHandler(errHandler))
c.Assert(err, IsNil)
proxy := httptest.NewServer(st)
@@ -257,7 +256,7 @@ func (s *STSuite) TestNotModified(c *C) {
})
// stream handler will forward requests to redirect
st, err := New(rdr, Logger(utils.NewFileLogger(os.Stdout, utils.INFO)))
st, err := New(rdr)
c.Assert(err, IsNil)
proxy := httptest.NewServer(st)
@@ -285,7 +284,7 @@ func (s *STSuite) TestNoBody(c *C) {
})
// stream handler will forward requests to redirect
st, err := New(rdr, Logger(utils.NewFileLogger(os.Stdout, utils.INFO)))
st, err := New(rdr)
c.Assert(err, IsNil)
proxy := httptest.NewServer(st)
@@ -317,7 +316,7 @@ func (s *STSuite) TestPreservesTLS(c *C) {
})
// stream handler will forward requests to redirect
st, err := New(rdr, Logger(utils.NewFileLogger(os.Stdout, utils.INFO)))
st, err := New(rdr)
c.Assert(err, IsNil)
proxy := httptest.NewUnstartedServer(st)
stream/threshold.go
-2
View File
@@ -4,7 +4,6 @@ import (
"fmt"
"net/http"
"github.com/vulcand/oxy/utils"
"github.com/vulcand/predicate"
)
@@ -17,7 +16,6 @@ type context struct {
r *http.Request
attempt int
responseCode int
log utils.Logger
}
type hpredicate func(*context) bool
trace/trace.go
+2 -13
View File
@@ -10,6 +10,7 @@ import (
"strconv"
"time"
log "github.com/Sirupsen/logrus"
"github.com/vulcand/oxy/utils"
)
@@ -40,14 +41,6 @@ func ResponseHeaders(headers ...string) Option {
}
}
// 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
@@ -55,7 +48,6 @@ type Tracer struct {
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
@@ -74,9 +66,6 @@ func New(next http.Handler, writer io.Writer, opts ...Option) (*Tracer, error) {
if t.errHandler == nil {
t.errHandler = utils.DefaultHandler
}
if t.log == nil {
t.log = utils.NullLogger
}
return t, nil
}
@@ -87,7 +76,7 @@ func (t *Tracer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
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)
log.Errorf("Failed to marshal request: %v", err)
}
}
trace/trace_test.go
+3 -12
View File
@@ -28,12 +28,9 @@ func (s *TraceSuite) TestTraceSimple(c *C) {
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))
t, err := New(handler, trace)
c.Assert(err, IsNil)
srv := httptest.NewServer(t)
@@ -64,12 +61,9 @@ func (s *TraceSuite) TestTraceCaptureHeaders(c *C) {
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"))
t, err := New(handler, trace, RequestHeaders("X-Req-B", "X-Req-A"), ResponseHeaders("X-Re-1", "X-Re-2"))
c.Assert(err, IsNil)
srv := httptest.NewServer(t)
@@ -91,12 +85,9 @@ 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))
t, err := New(handler, trace)
c.Assert(err, IsNil)
srv := httptest.NewUnstartedServer(t)
utils/logging.go
-86
View File
@@ -1,86 +0,0 @@
package utils
import (
"io"
"log"
)
var NullLogger Logger = &NOPLogger{}
// Logger defines a simple logging interface
type Logger interface {
Infof(format string, args ...interface{})
Warningf(format string, args ...interface{})
Errorf(format string, args ...interface{})
}
type FileLogger struct {
info *log.Logger
warn *log.Logger
error *log.Logger
}
func NewFileLogger(w io.Writer, lvl LogLevel) *FileLogger {
l := &FileLogger{}
flag := log.Ldate | log.Ltime | log.Lmicroseconds
if lvl <= INFO {
l.info = log.New(w, "INFO: ", flag)
}
if lvl <= WARN {
l.warn = log.New(w, "WARN: ", flag)
}
if lvl <= ERROR {
l.error = log.New(w, "ERR: ", flag)
}
return l
}
func (f *FileLogger) Infof(format string, args ...interface{}) {
if f.info == nil {
return
}
f.info.Printf(format, args...)
}
func (f *FileLogger) Warningf(format string, args ...interface{}) {
if f.warn == nil {
return
}
f.warn.Printf(format, args...)
}
func (f *FileLogger) Errorf(format string, args ...interface{}) {
if f.error == nil {
return
}
f.error.Printf(format, args...)
}
type NOPLogger struct {
}
func (*NOPLogger) Infof(format string, args ...interface{}) {
}
func (*NOPLogger) Warningf(format string, args ...interface{}) {
}
func (*NOPLogger) Errorf(format string, args ...interface{}) {
}
func (*NOPLogger) Info(string) {
}
func (*NOPLogger) Warning(string) {
}
func (*NOPLogger) Error(string) {
}
type LogLevel int
const (
INFO = iota
WARN
ERROR
)