diff --git a/go.sum b/go.sum index 9031a6131..b330071a0 100644 --- a/go.sum +++ b/go.sum @@ -5,6 +5,7 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pty v1.1.1 h1:VkoXIwSboBpnk99O/KFauAEILuNHv5DVFKZMBN/gUgw= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= @@ -12,6 +13,7 @@ github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0 h1:4G4v2dO3VZwixGIRoQ5Lfboy6nUhCyYzaqnIAPPhYs4= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= @@ -23,17 +25,23 @@ go.uber.org/goleak v1.1.10 h1:z+mqJhf6ss6BSfSM671tgKyZBFPTTJM+HLxnhPC3wu0= go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A= go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4= go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2 h1:VklqNMn3ovrHsnt90PveolxSbWFaJdECFbxSq0Mqo2M= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/lint v0.0.0-20190930215403-16217165b5de h1:5hukYrvBGR8/eNkX5mdUezrA6JiaEZDtJb9Ei+1LlBs= golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859 h1:R/3boaszxrf1GEUWTVDzSKVwLmSJpwZ1yqXm8j0v2QI= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/sync v0.0.0-20190423024810-112230192c58 h1:8gQV6CLnAEikrhgkHFbMAEhagSSnXWGV915qUMm9mrU= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a h1:1BGLXjeY4akVXGgbC9HugT3Jv3hCI0z56oJR5vAMgBU= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= golang.org/x/tools v0.0.0-20191108193012-7d206e10da11 h1:Yq9t9jnGoR+dBuitxdo9l6Q7xh/zOyNnYUtDKaQ3x0E= golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7 h1:9zdDQZ7Thm29KFXgAX/+yaf3eVbP7djjWp/dXAppNCc= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY= diff --git a/zapcore/buffered_write_syncer.go b/zapcore/buffered_write_syncer.go new file mode 100644 index 000000000..f08ca7e81 --- /dev/null +++ b/zapcore/buffered_write_syncer.go @@ -0,0 +1,176 @@ +// Copyright (c) 2021 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package zapcore + +import ( + "bufio" + "sync" + "time" + + "go.uber.org/multierr" +) + +const ( + // _defaultBufferSize specifies the default size used by Buffer. + _defaultBufferSize = 256 * 1024 // 256 kB + + // _defaultFlushInterval specifies the default flush interval for + // Buffer. + _defaultFlushInterval = 30 * time.Second +) + +// A BufferedWriteSyncer is a WriteSyncer that buffers writes in-memory before +// flushing them to a wrapped WriteSyncer after reaching some limit, or at some +// fixed interval--whichever comes first. +// +// BufferedWriteSyncer is safe for concurrent use. You don't need to use +// zapcore.Lock for WriteSyncers with BufferedWriteSyncer. +type BufferedWriteSyncer struct { + // WS is the WriteSyncer around which BufferedWriteSyncer will buffer + // writes. + // + // This field is required. + WS WriteSyncer + + // Size specifies the maximum amount of data the writer will buffered + // before flushing. + // + // Defaults to 256 kB if unspecified. + Size int + + // FlushInterval specifies how often the writer should flush data if + // there have been no writes. + // + // Defaults to 30 seconds if unspecified. + FlushInterval time.Duration + + // Clock, if specified, provides control of the source of time for the + // writer. + // + // Defaults to the system clock. + Clock Clock + + // unexported fields for state + mu sync.Mutex + initialized bool // whether initialize() has run + writer *bufio.Writer + ticker *time.Ticker + stop chan struct{} // closed when flushLoop should stop + done chan struct{} // closed when flushLoop has stopped +} + +func (s *BufferedWriteSyncer) initialize() { + size := s.Size + if size == 0 { + size = _defaultBufferSize + } + + flushInterval := s.FlushInterval + if flushInterval == 0 { + flushInterval = _defaultFlushInterval + } + + if s.Clock == nil { + s.Clock = DefaultClock + } + + s.ticker = s.Clock.NewTicker(flushInterval) + s.writer = bufio.NewWriterSize(s.WS, size) + s.stop = make(chan struct{}) + s.done = make(chan struct{}) + s.initialized = true + go s.flushLoop() +} + +// Write writes log data into buffer syncer directly, multiple Write calls will be batched, +// and log data will be flushed to disk when the buffer is full or periodically. +func (s *BufferedWriteSyncer) Write(bs []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if !s.initialized { + s.initialize() + } + + // To avoid partial writes from being flushed, we manually flush the existing buffer if: + // * The current write doesn't fit into the buffer fully, and + // * The buffer is not empty (since bufio will not split large writes when the buffer is empty) + if len(bs) > s.writer.Available() && s.writer.Buffered() > 0 { + if err := s.writer.Flush(); err != nil { + return 0, err + } + } + + return s.writer.Write(bs) +} + +// Sync flushes buffered log data into disk directly. +func (s *BufferedWriteSyncer) Sync() error { + s.mu.Lock() + defer s.mu.Unlock() + + var err error + if s.initialized { + err = s.writer.Flush() + } + + return multierr.Append(err, s.WS.Sync()) +} + +// flushLoop flushes the buffer at the configured interval until Stop is +// called. +func (s *BufferedWriteSyncer) flushLoop() { + defer close(s.done) + + for { + select { + case <-s.ticker.C: + // we just simply ignore error here + // because the underlying bufio writer stores any errors + // and we return any error from Sync() as part of the close + _ = s.Sync() + case <-s.stop: + return + } + } +} + +// Stop closes the buffer, cleans up background goroutines, and flushes +// remaining unwritten data. +// +// This must be called exactly once. +func (s *BufferedWriteSyncer) Stop() error { + // Critical section. + func() { + s.mu.Lock() + defer s.mu.Unlock() + + if !s.initialized { + return + } + + s.ticker.Stop() + close(s.stop) // tell flushLoop to stop + <-s.done // and wait until it has + }() + + return s.Sync() +} diff --git a/zapcore/buffered_write_syncer_bench_test.go b/zapcore/buffered_write_syncer_bench_test.go new file mode 100644 index 000000000..dd1583c14 --- /dev/null +++ b/zapcore/buffered_write_syncer_bench_test.go @@ -0,0 +1,53 @@ +// Copyright (c) 2021 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package zapcore + +import ( + "io/ioutil" + "os" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func BenchmarkBufferedWriteSyncer(b *testing.B) { + b.Run("write file with buffer", func(b *testing.B) { + file, err := ioutil.TempFile("", "log") + require.NoError(b, err) + + defer func() { + assert.NoError(b, file.Close()) + assert.NoError(b, os.Remove(file.Name())) + }() + + w := &BufferedWriteSyncer{ + WS: AddSync(file), + } + defer w.Stop() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + w.Write([]byte("foobarbazbabble")) + } + }) + }) +} diff --git a/zapcore/buffered_write_syncer_test.go b/zapcore/buffered_write_syncer_test.go new file mode 100644 index 000000000..9c28076bb --- /dev/null +++ b/zapcore/buffered_write_syncer_test.go @@ -0,0 +1,131 @@ +// Copyright (c) 2021 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package zapcore + +import ( + "bytes" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap/internal/ztest" +) + +func TestBufferWriter(t *testing.T) { + // If we pass a plain io.Writer, make sure that we still get a WriteSyncer + // with a no-op Sync. + t.Run("sync", func(t *testing.T) { + buf := &bytes.Buffer{} + ws := &BufferedWriteSyncer{WS: AddSync(buf)} + + requireWriteWorks(t, ws) + assert.Empty(t, buf.String(), "Unexpected log calling a no-op Write method.") + assert.NoError(t, ws.Sync(), "Unexpected error calling a no-op Sync method.") + assert.Equal(t, "foo", buf.String(), "Unexpected log string") + assert.NoError(t, ws.Stop()) + }) + + t.Run("stop", func(t *testing.T) { + buf := &bytes.Buffer{} + ws := &BufferedWriteSyncer{WS: AddSync(buf)} + requireWriteWorks(t, ws) + assert.Empty(t, buf.String(), "Unexpected log calling a no-op Write method.") + assert.NoError(t, ws.Stop()) + assert.Equal(t, "foo", buf.String(), "Unexpected log string") + }) + + t.Run("wrap twice", func(t *testing.T) { + buf := &bytes.Buffer{} + bufsync := &BufferedWriteSyncer{WS: AddSync(buf)} + ws := &BufferedWriteSyncer{WS: bufsync} + requireWriteWorks(t, ws) + assert.Empty(t, buf.String(), "Unexpected log calling a no-op Write method.") + require.NoError(t, ws.Sync()) + assert.Equal(t, "foo", buf.String()) + assert.NoError(t, ws.Stop()) + assert.NoError(t, bufsync.Stop()) + assert.Equal(t, "foo", buf.String(), "Unexpected log string") + }) + + t.Run("small buffer", func(t *testing.T) { + buf := &bytes.Buffer{} + ws := &BufferedWriteSyncer{WS: AddSync(buf), Size: 5} + + requireWriteWorks(t, ws) + assert.Equal(t, "", buf.String(), "Unexpected log calling a no-op Write method.") + requireWriteWorks(t, ws) + assert.Equal(t, "foo", buf.String(), "Unexpected log string") + assert.NoError(t, ws.Stop()) + }) + + t.Run("with lockedWriteSyncer", func(t *testing.T) { + buf := &bytes.Buffer{} + ws := &BufferedWriteSyncer{WS: Lock(AddSync(buf)), Size: 5} + + requireWriteWorks(t, ws) + assert.Equal(t, "", buf.String(), "Unexpected log calling a no-op Write method.") + requireWriteWorks(t, ws) + assert.Equal(t, "foo", buf.String(), "Unexpected log string") + assert.NoError(t, ws.Stop()) + }) + + t.Run("flush error", func(t *testing.T) { + ws := &BufferedWriteSyncer{WS: &ztest.FailWriter{}, Size: 4} + n, err := ws.Write([]byte("foo")) + require.NoError(t, err, "Unexpected error writing to WriteSyncer.") + require.Equal(t, 3, n, "Wrote an unexpected number of bytes.") + ws.Write([]byte("foo")) + assert.Error(t, ws.Stop(), "Expected stop to fail.") + }) + + t.Run("flush timer", func(t *testing.T) { + buf := &bytes.Buffer{} + clock := newControlledClock() + ws := &BufferedWriteSyncer{ + WS: AddSync(buf), + Size: 6, + FlushInterval: time.Microsecond, + Clock: clock, + } + requireWriteWorks(t, ws) + clock.Add(10 * time.Microsecond) + assert.Equal(t, "foo", buf.String(), "Unexpected log string") + + // flush twice to validate loop logic + requireWriteWorks(t, ws) + clock.Add(10 * time.Microsecond) + assert.Equal(t, "foofoo", buf.String(), "Unexpected log string") + assert.NoError(t, ws.Stop()) + }) +} + +func TestBufferWriterWithoutStart(t *testing.T) { + t.Run("stop", func(t *testing.T) { + ws := &BufferedWriteSyncer{WS: AddSync(new(bytes.Buffer))} + assert.NoError(t, ws.Stop(), "Stop must not fail") + }) + + t.Run("Sync", func(t *testing.T) { + ws := &BufferedWriteSyncer{WS: AddSync(new(bytes.Buffer))} + assert.NoError(t, ws.Sync(), "Sync must not fail") + }) +} diff --git a/zapcore/write_syncer_bench_test.go b/zapcore/write_syncer_bench_test.go index 0209d0f61..0793805d5 100644 --- a/zapcore/write_syncer_bench_test.go +++ b/zapcore/write_syncer_bench_test.go @@ -21,13 +21,16 @@ package zapcore import ( + "io/ioutil" + "os" "testing" + "github.com/stretchr/testify/assert" "go.uber.org/zap/internal/ztest" ) func BenchmarkMultiWriteSyncer(b *testing.B) { - b.Run("2", func(b *testing.B) { + b.Run("2 discarder", func(b *testing.B) { w := NewMultiWriteSyncer( &ztest.Discarder{}, &ztest.Discarder{}, @@ -39,7 +42,7 @@ func BenchmarkMultiWriteSyncer(b *testing.B) { } }) }) - b.Run("4", func(b *testing.B) { + b.Run("4 discarder", func(b *testing.B) { w := NewMultiWriteSyncer( &ztest.Discarder{}, &ztest.Discarder{}, @@ -53,4 +56,38 @@ func BenchmarkMultiWriteSyncer(b *testing.B) { } }) }) + b.Run("4 discarder with buffer", func(b *testing.B) { + w := &BufferedWriteSyncer{ + WS: NewMultiWriteSyncer( + &ztest.Discarder{}, + &ztest.Discarder{}, + &ztest.Discarder{}, + &ztest.Discarder{}, + ), + } + defer w.Stop() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + w.Write([]byte("foobarbazbabble")) + } + }) + }) +} + +func BenchmarkWriteSyncer(b *testing.B) { + b.Run("write file with no buffer", func(b *testing.B) { + file, err := ioutil.TempFile("", "log") + assert.NoError(b, err) + defer file.Close() + defer os.Remove(file.Name()) + + w := AddSync(file) + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + w.Write([]byte("foobarbazbabble")) + } + }) + }) } diff --git a/zapcore/write_syncer_test.go b/zapcore/write_syncer_test.go index 3ccb0af24..4748be7f5 100644 --- a/zapcore/write_syncer_test.go +++ b/zapcore/write_syncer_test.go @@ -23,9 +23,8 @@ package zapcore import ( "bytes" "errors" - "testing" - "io" + "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require"