Merge remote-tracking branch 'origin/disk-buffer' into submod-diskbuff

operator/buffer/buffer.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"time"
 
 	"github.com/observiq/stanza/entry"
 	"github.com/observiq/stanza/operator"
@@ -13,7 +12,7 @@ import (
 type Buffer interface {
 	Add(context.Context, *entry.Entry) error
 	Read([]*entry.Entry) (func(), int, error)
-	ReadWait([]*entry.Entry, <-chan time.Time) (func(), int, error)
+	ReadWait(context.Context, []*entry.Entry) (func(), int, error)
 	Close() error
 }
 

operator/buffer/buffer_test.go

@@ -31,6 +31,7 @@ func TestBufferUnmarshalYAML(t *testing.T) {
 				BufferBuilder: &DiskBufferConfig{
 					MaxSize: 1234,
 					Path:    "/var/log/testpath",
+					Sync:    true,
 				},
 			},
 		},

operator/buffer/disk.go

@@ -20,18 +20,20 @@ type DiskBufferConfig struct {
 	// TODO make this configurable in human-readable terms
 	MaxSize int    `json:"max_size" yaml:"max_size"`
 	Path    string `json:"path" yaml:"path"`
+	Sync    bool   `json:"sync" yaml:"sync"`
 }
 
 // NewDiskBufferConfig creates a new default disk buffer config
 func NewDiskBufferConfig() *DiskBufferConfig {
 	return &DiskBufferConfig{
 		MaxSize: 1 << 32, // 4GiB
+		Sync:    true,
 	}
 }
 
 func (c DiskBufferConfig) Build(context operator.BuildContext, _ string) (Buffer, error) {
 	b := NewDiskBuffer(c.MaxSize)
-	if err := b.Open(c.Path); err != nil {
+	if err := b.Open(c.Path, c.Sync); err != nil {
 		return nil, err
 	}
 	return b, nil
@@ -81,15 +83,19 @@ func NewDiskBuffer(maxDiskSize int) *DiskBuffer {
 }
 
 // Open opens the disk buffer files from a database directory
-func (d *DiskBuffer) Open(path string) error {
+func (d *DiskBuffer) Open(path string, sync bool) error {
 	var err error
 	dataPath := filepath.Join(path, "data")
-	if d.data, err = os.OpenFile(dataPath, os.O_CREATE|os.O_RDWR, 0755); err != nil {
+	flags := os.O_CREATE | os.O_RDWR
+	if sync {
+		flags |= os.O_SYNC
+	}
+	if d.data, err = os.OpenFile(dataPath, flags, 0755); err != nil {
 		return err
 	}
 
 	metadataPath := filepath.Join(path, "metadata")
-	if d.metadata, err = OpenMetadata(metadataPath); err != nil {
+	if d.metadata, err = OpenMetadata(metadataPath, sync); err != nil {
 		return err
 	}
 
@@ -178,10 +184,10 @@ func (d *DiskBuffer) addUnreadCount(i int64) {
 }
 
 // ReadWait reads entries from the buffer, waiting until either there are enough entries in the
-// buffer to fill dst, or an event is sent down the timeout channel. This amortizes the cost
-// of reading from the disk. It returns a function that, when called, marks the read entries as
-// flushed, the number of entries read, and an error.
-func (d *DiskBuffer) ReadWait(dst []*entry.Entry, timeout <-chan time.Time) (func(), int, error) {
+// buffer to fill dst or the context is cancelled. This amortizes the cost of reading from the
+// disk. It returns a function that, when called, marks the read entries as flushed, the
+// number of entries read, and an error.
+func (d *DiskBuffer) ReadWait(ctx context.Context, dst []*entry.Entry) (func(), int, error) {
 	d.readerLock.Lock()
 	defer d.readerLock.Unlock()
 
@@ -193,7 +199,7 @@ LOOP:
 			if n >= int64(len(dst)) {
 				break LOOP
 			}
-		case <-timeout:
+		case <-ctx.Done():
 			break LOOP
 		}
 	}

operator/buffer/disk_metadata.go

@@ -45,11 +45,15 @@ type Metadata struct {
 }
 
 // OpenMetadata opens and parses the metadata
-func OpenMetadata(path string) (*Metadata, error) {
+func OpenMetadata(path string, sync bool) (*Metadata, error) {
 	m := &Metadata{}
 
 	var err error
-	if m.file, err = os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0755); err != nil {
+	flags := os.O_CREATE | os.O_RDWR
+	if sync {
+		flags |= os.O_SYNC
+	}
+	if m.file, err = os.OpenFile(path, flags, 0755); err != nil {
 		return &Metadata{}, err
 	}
 

operator/buffer/disk_test.go

@@ -18,7 +18,7 @@ import (
 func openBuffer(t testing.TB) *DiskBuffer {
 	buffer := NewDiskBuffer(1 << 20)
 	dir := testutil.NewTempDir(t)
-	err := buffer.Open(dir)
+	err := buffer.Open(dir, false)
 	require.NoError(t, err)
 	t.Cleanup(func() { buffer.Close() })
 	return buffer
@@ -130,7 +130,7 @@ func TestDiskBuffer(t *testing.T) {
 		t.Parallel()
 		b := NewDiskBuffer(1 << 30)
 		dir := testutil.NewTempDir(t)
-		err := b.Open(dir)
+		err := b.Open(dir, false)
 		require.NoError(t, err)
 
 		writeN(t, b, 20, 0)
@@ -139,7 +139,7 @@ func TestDiskBuffer(t *testing.T) {
 		require.NoError(t, err)
 
 		b2 := NewDiskBuffer(1 << 30)
-		err = b2.Open(dir)
+		err = b2.Open(dir, false)
 		require.NoError(t, err)
 		readN(t, b2, 20, 0)
 	})
@@ -148,7 +148,7 @@ func TestDiskBuffer(t *testing.T) {
 		t.Parallel()
 		b := NewDiskBuffer(1 << 30)
 		dir := testutil.NewTempDir(t)
-		err := b.Open(dir)
+		err := b.Open(dir, false)
 		require.NoError(t, err)
 
 		writeN(t, b, 20, 0)
@@ -157,7 +157,7 @@ func TestDiskBuffer(t *testing.T) {
 		require.NoError(t, err)
 
 		b2 := NewDiskBuffer(1 << 30)
-		err = b2.Open(dir)
+		err = b2.Open(dir, false)
 		require.NoError(t, err)
 		readN(t, b2, 10, 10)
 	})
@@ -173,7 +173,7 @@ func TestDiskBuffer(t *testing.T) {
 
 				b := NewDiskBuffer(1 << 30)
 				dir := testutil.NewTempDir(t)
-				err := b.Open(dir)
+				err := b.Open(dir, false)
 				require.NoError(t, err)
 
 				writes := 0
@@ -204,32 +204,75 @@ func TestDiskBuffer(t *testing.T) {
 }
 
 func BenchmarkDiskBuffer(b *testing.B) {
-	buffer := openBuffer(b)
-	var wg sync.WaitGroup
-
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		fmt.Printf("Benchmark: %d\n", b.N)
-		e := entry.New()
-		e.Record = "test log"
-		ctx := context.Background()
-		for i := 0; i < b.N; i++ {
-			panicOnErr(buffer.Add(ctx, e))
-		}
-	}()
-
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		dst := make([]*entry.Entry, 1000)
-		for i := 0; i < b.N; {
-			flush, n, err := buffer.ReadWait(dst, time.After(50*time.Millisecond))
-			panicOnErr(err)
-			i += n
-			flush()
-		}
-	}()
+	b.Run("NoSync", func(b *testing.B) {
+		buffer := openBuffer(b)
+		var wg sync.WaitGroup
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			fmt.Printf("Benchmark: %d\n", b.N)
+			e := entry.New()
+			e.Record = "test log"
+			ctx := context.Background()
+			for i := 0; i < b.N; i++ {
+				panicOnErr(buffer.Add(ctx, e))
+			}
+		}()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			dst := make([]*entry.Entry, 1000)
+			ctx := context.Background()
+			for i := 0; i < b.N; {
+				ctx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
+				flush, n, err := buffer.ReadWait(ctx, dst)
+				cancel()
+				panicOnErr(err)
+				i += n
+				flush()
+			}
+		}()
+
+		wg.Wait()
+	})
+
+	b.Run("Sync", func(b *testing.B) {
+		buffer := NewDiskBuffer(1 << 20)
+		dir := testutil.NewTempDir(b)
+		err := buffer.Open(dir, true)
+		require.NoError(b, err)
+		b.Cleanup(func() { buffer.Close() })
+		var wg sync.WaitGroup
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			fmt.Printf("Benchmark: %d\n", b.N)
+			e := entry.New()
+			e.Record = "test log"
+			ctx := context.Background()
+			for i := 0; i < b.N; i++ {
+				panicOnErr(buffer.Add(ctx, e))
+			}
+		}()
 
-	wg.Wait()
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			dst := make([]*entry.Entry, 1000)
+			ctx := context.Background()
+			for i := 0; i < b.N; {
+				ctx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
+				flush, n, err := buffer.ReadWait(ctx, dst)
+				cancel()
+				panicOnErr(err)
+				i += n
+				flush()
+			}
+		}()
+
+		wg.Wait()
+	})
 }

operator/buffer/memory.go

@@ -4,11 +4,10 @@ import (
 	"bytes"
 	"context"
 	"encoding/binary"
-	"encoding/gob"
+	"encoding/json"
 	"fmt"
 	"sync"
 	"sync/atomic"
-	"time"
 
 	"github.com/observiq/stanza/entry"
 	"github.com/observiq/stanza/operator"
@@ -60,7 +59,7 @@ func (m *MemoryBuffer) Add(ctx context.Context, e *entry.Entry) error {
 	case m.buf <- e:
 		return nil
 	case <-ctx.Done():
-		return fmt.Errorf("context cancelled")
+		return ctx.Err()
 	}
 }
 
@@ -84,7 +83,7 @@ func (m *MemoryBuffer) Read(dst []*entry.Entry) (func(), int, error) {
 	return m.newFlushFunc(inFlight[:i]), i, nil
 }
 
-func (m *MemoryBuffer) ReadWait(dst []*entry.Entry, timeout <-chan time.Time) (func(), int, error) {
+func (m *MemoryBuffer) ReadWait(ctx context.Context, dst []*entry.Entry) (func(), int, error) {
 	inFlightIDs := make([]uint64, len(dst))
 	i := 0
 	for ; i < len(dst); i++ {
@@ -96,7 +95,7 @@ func (m *MemoryBuffer) ReadWait(dst []*entry.Entry, timeout <-chan time.Time) (f
 			m.inFlight[id] = e
 			m.inFlightMux.Unlock()
 			inFlightIDs[i] = id
-		case <-timeout:
+		case <-ctx.Done():
 			return m.newFlushFunc(inFlightIDs[:i]), i, nil
 		}
 	}
@@ -133,21 +132,23 @@ func (m *MemoryBuffer) Close() error {
 			}
 		}
 
-		close(m.buf)
-		for e := range m.buf {
-			m.entryID++
-			if err := putKeyValue(b, m.entryID, e); err != nil {
-				return err
+		for {
+			select {
+			case e := <-m.buf:
+				m.entryID++
+				if err := putKeyValue(b, m.entryID, e); err != nil {
+					return err
+				}
+			default:
+				return nil
 			}
 		}
-
-		return nil
 	})
 }
 
 func putKeyValue(b *bbolt.Bucket, k uint64, v *entry.Entry) error {
 	var buf bytes.Buffer
-	enc := gob.NewEncoder(&buf)
+	enc := json.NewEncoder(&buf)
 	key := [8]byte{}
 
 	binary.LittleEndian.PutUint64(key[:], k)
@@ -174,7 +175,7 @@ func (m *MemoryBuffer) loadFromDB() error {
 				return fmt.Errorf("max_entries is smaller than the number of entries stored in the database")
 			}
 
-			dec := gob.NewDecoder(bytes.NewReader(v))
+			dec := json.NewDecoder(bytes.NewReader(v))
 			var e entry.Entry
 			if err := dec.Decode(&e); err != nil {
 				return err

operator/buffer/memory_test.go

@@ -44,6 +44,28 @@ func TestMemoryBuffer(t *testing.T) {
 		readN(t, b, 10, 10)
 	})
 
+	t.Run("CheckN", func(t *testing.T) {
+		t.Run("Read", func(t *testing.T) {
+			b := newMemoryBuffer(t)
+			writeN(t, b, 12, 0)
+			dst := make([]*entry.Entry, 30)
+			_, n, err := b.Read(dst)
+			require.NoError(t, err)
+			require.Equal(t, n, 12)
+		})
+
+		t.Run("ReadWait", func(t *testing.T) {
+			b := newMemoryBuffer(t)
+			writeN(t, b, 12, 0)
+			dst := make([]*entry.Entry, 30)
+			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
+			defer cancel()
+			_, n, err := b.ReadWait(ctx, dst)
+			require.NoError(t, err)
+			require.Equal(t, n, 12)
+		})
+	})
+
 	t.Run("SingleReadWaitMultipleWrites", func(t *testing.T) {
 		t.Parallel()
 		b := newMemoryBuffer(t)
@@ -173,8 +195,11 @@ func BenchmarkMemoryBuffer(b *testing.B) {
 	go func() {
 		defer wg.Done()
 		dst := make([]*entry.Entry, 1000)
+		ctx := context.Background()
 		for i := 0; i < b.N; {
-			flush, n, err := buffer.ReadWait(dst, time.After(50*time.Millisecond))
+			ctx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
+			flush, n, err := buffer.ReadWait(ctx, dst)
+			cancel()
 			panicOnErr(err)
 			i += n
 			go func() {