pre- and post-snapshot hooks
* stack-based execution model, documented in documentation * circbuf for capturing hook output * built-in hooks for postgres and mysql * refactor docs, too much info on the jobs page, too difficult to discover snapshotting & hooks Co-authored-by: Ross Williams <ross@ross-williams.net> Co-authored-by: Christian Schwarz <me@cschwarz.com> fixes #74
This commit is contained in:
Ross Williams
committed by
Christian Schwarz
Christian Schwarz
parent
00434f4ac9
commit
729c83ee72
@@ -0,0 +1,167 @@
|
||||
package circlog
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math/bits"
|
||||
"sync"
|
||||
)
|
||||
|
||||
const CIRCULARLOG_INIT_SIZE int = 32 << 10
|
||||
|
||||
type CircularLog struct {
|
||||
buf []byte
|
||||
size int
|
||||
max int
|
||||
written int
|
||||
writeCursor int
|
||||
/*
|
||||
Mutex prevents:
|
||||
concurrent writes:
|
||||
buf, size, written, writeCursor in Write([]byte)
|
||||
buf, writeCursor in Reset()
|
||||
data races vs concurrent Write([]byte) calls:
|
||||
size in Size()
|
||||
size, writeCursor in Len()
|
||||
buf, size, written, writeCursor in Bytes()
|
||||
*/
|
||||
mtx sync.Mutex
|
||||
}
|
||||
|
||||
func NewCircularLog(max int) (*CircularLog, error) {
|
||||
if max <= 0 {
|
||||
return nil, fmt.Errorf("max must be positive")
|
||||
}
|
||||
|
||||
return &CircularLog{
|
||||
size: CIRCULARLOG_INIT_SIZE,
|
||||
buf: make([]byte, CIRCULARLOG_INIT_SIZE),
|
||||
max: max,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func nextPow2Int(n int) int {
|
||||
if n < 1 {
|
||||
panic("can only round up positive integers")
|
||||
}
|
||||
r := uint(n)
|
||||
// Credit: https://graphics.stanford.edu/~seander/bithacks.html#RoundUpPowerOf2
|
||||
r--
|
||||
// Assume at least a 32 bit integer
|
||||
r |= r >> 1
|
||||
r |= r >> 2
|
||||
r |= r >> 4
|
||||
r |= r >> 8
|
||||
r |= r >> 16
|
||||
if bits.UintSize == 64 {
|
||||
r |= r >> 32
|
||||
}
|
||||
r++
|
||||
// Can't exceed max positive int value
|
||||
if r > ^uint(0)>>1 {
|
||||
panic("rounded to larger than int()")
|
||||
}
|
||||
return int(r)
|
||||
}
|
||||
|
||||
func (cl *CircularLog) Write(data []byte) (int, error) {
|
||||
cl.mtx.Lock()
|
||||
defer cl.mtx.Unlock()
|
||||
n := len(data)
|
||||
|
||||
// Keep growing the buffer by doubling until
|
||||
// hitting cl.max
|
||||
bufAvail := cl.size - cl.writeCursor
|
||||
// If cl.writeCursor wrapped on the last write,
|
||||
// then the buffer is full, not empty.
|
||||
if cl.writeCursor == 0 && cl.written > 0 {
|
||||
bufAvail = 0
|
||||
}
|
||||
if n > bufAvail && cl.size < cl.max {
|
||||
// Add to size, not writeCursor, so as
|
||||
// to not resize multiple times if this
|
||||
// Write() immediately fills up the buffer
|
||||
newSize := nextPow2Int(cl.size + n)
|
||||
if newSize > cl.max {
|
||||
newSize = cl.max
|
||||
}
|
||||
newBuf := make([]byte, newSize)
|
||||
// Reset write cursor to old size if wrapped
|
||||
if cl.writeCursor == 0 && cl.written > 0 {
|
||||
cl.writeCursor = cl.size
|
||||
}
|
||||
copy(newBuf, cl.buf[:cl.writeCursor])
|
||||
cl.buf = newBuf
|
||||
cl.size = newSize
|
||||
}
|
||||
|
||||
// If data to be written is larger than the max size,
|
||||
// discard all but the last cl.size bytes
|
||||
if n > cl.size {
|
||||
data = data[n-cl.size:]
|
||||
// Overwrite the beginning of data
|
||||
// with a string indicating truncation
|
||||
copy(data, []byte("(...)"))
|
||||
}
|
||||
// First copy data to the end of buf. If that wasn't
|
||||
// all of data, then copy the rest to the beginning
|
||||
// of buf.
|
||||
copied := copy(cl.buf[cl.writeCursor:], data)
|
||||
if copied < n {
|
||||
copied += copy(cl.buf, data[copied:])
|
||||
}
|
||||
|
||||
cl.writeCursor = ((cl.writeCursor + copied) % cl.size)
|
||||
cl.written += copied
|
||||
return copied, nil
|
||||
}
|
||||
|
||||
func (cl *CircularLog) Size() int {
|
||||
cl.mtx.Lock()
|
||||
defer cl.mtx.Unlock()
|
||||
return cl.size
|
||||
}
|
||||
|
||||
func (cl *CircularLog) Len() int {
|
||||
cl.mtx.Lock()
|
||||
defer cl.mtx.Unlock()
|
||||
if cl.written >= cl.size {
|
||||
return cl.size
|
||||
} else {
|
||||
return cl.writeCursor
|
||||
}
|
||||
}
|
||||
|
||||
func (cl *CircularLog) TotalWritten() int {
|
||||
cl.mtx.Lock()
|
||||
defer cl.mtx.Unlock()
|
||||
|
||||
return cl.written
|
||||
}
|
||||
|
||||
func (cl *CircularLog) Reset() {
|
||||
cl.mtx.Lock()
|
||||
defer cl.mtx.Unlock()
|
||||
cl.writeCursor = 0
|
||||
cl.buf = make([]byte, CIRCULARLOG_INIT_SIZE)
|
||||
cl.size = CIRCULARLOG_INIT_SIZE
|
||||
}
|
||||
|
||||
func (cl *CircularLog) Bytes() []byte {
|
||||
cl.mtx.Lock()
|
||||
defer cl.mtx.Unlock()
|
||||
switch {
|
||||
case cl.written >= cl.size && cl.writeCursor == 0:
|
||||
return cl.buf
|
||||
case cl.written > cl.size:
|
||||
ret := make([]byte, cl.size)
|
||||
copy(ret, cl.buf[cl.writeCursor:])
|
||||
copy(ret[cl.size-cl.writeCursor:], cl.buf[:cl.writeCursor])
|
||||
return ret
|
||||
default:
|
||||
return cl.buf[:cl.writeCursor]
|
||||
}
|
||||
}
|
||||
|
||||
func (cl *CircularLog) String() string {
|
||||
return string(cl.Bytes())
|
||||
}
|
||||
@@ -0,0 +1,144 @@
|
||||
package circlog_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"io"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/zrepl/zrepl/util/circlog"
|
||||
)
|
||||
|
||||
func TestCircularLog(t *testing.T) {
|
||||
var maxCircularLogSize int = 1 << 20
|
||||
|
||||
writeFixedSize := func(w io.Writer, size int) (int, error) {
|
||||
pattern := []byte{0xDE, 0xAD, 0xBE, 0xEF}
|
||||
writeBytes := bytes.Repeat(pattern, size/4)
|
||||
if len(writeBytes) < size {
|
||||
writeBytes = append(writeBytes, pattern[:size-len(writeBytes)]...)
|
||||
}
|
||||
n, err := w.Write(writeBytes)
|
||||
if err != nil {
|
||||
return n, err
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
t.Run("negative-size-error", func(t *testing.T) {
|
||||
_, err := circlog.NewCircularLog(-1)
|
||||
require.EqualError(t, err, "max must be positive")
|
||||
})
|
||||
|
||||
t.Run("no-resize", func(t *testing.T) {
|
||||
log, err := circlog.NewCircularLog(maxCircularLogSize)
|
||||
require.NoError(t, err)
|
||||
|
||||
initSize := log.Size()
|
||||
writeSize := circlog.CIRCULARLOG_INIT_SIZE - 1
|
||||
_, err = writeFixedSize(log, writeSize)
|
||||
require.NoError(t, err)
|
||||
finalSize := log.Size()
|
||||
require.Equal(t, initSize, finalSize)
|
||||
require.Equal(t, writeSize, log.Len())
|
||||
})
|
||||
t.Run("one-resize", func(t *testing.T) {
|
||||
log, err := circlog.NewCircularLog(maxCircularLogSize)
|
||||
require.NoError(t, err)
|
||||
|
||||
initSize := log.Size()
|
||||
writeSize := circlog.CIRCULARLOG_INIT_SIZE + 1
|
||||
_, err = writeFixedSize(log, writeSize)
|
||||
require.NoError(t, err)
|
||||
finalSize := log.Size()
|
||||
require.Greater(t, finalSize, initSize)
|
||||
require.LessOrEqual(t, finalSize, maxCircularLogSize)
|
||||
require.Equal(t, writeSize, log.Len())
|
||||
})
|
||||
t.Run("reset", func(t *testing.T) {
|
||||
log, err := circlog.NewCircularLog(maxCircularLogSize)
|
||||
require.NoError(t, err)
|
||||
|
||||
initSize := log.Size()
|
||||
_, err = writeFixedSize(log, maxCircularLogSize)
|
||||
require.NoError(t, err)
|
||||
log.Reset()
|
||||
finalSize := log.Size()
|
||||
require.Equal(t, initSize, finalSize)
|
||||
})
|
||||
t.Run("wrap-exactly-maximum", func(t *testing.T) {
|
||||
log, err := circlog.NewCircularLog(maxCircularLogSize)
|
||||
require.NoError(t, err)
|
||||
|
||||
initSize := log.Size()
|
||||
writeSize := maxCircularLogSize / 4
|
||||
for written := 0; written < maxCircularLogSize; {
|
||||
n, err := writeFixedSize(log, writeSize)
|
||||
written += n
|
||||
require.NoError(t, err)
|
||||
}
|
||||
finalSize := log.Size()
|
||||
require.Greater(t, finalSize, initSize)
|
||||
require.Equal(t, finalSize, maxCircularLogSize)
|
||||
require.Equal(t, finalSize, log.Len())
|
||||
})
|
||||
t.Run("wrap-partial", func(t *testing.T) {
|
||||
log, err := circlog.NewCircularLog(maxCircularLogSize)
|
||||
require.NoError(t, err)
|
||||
|
||||
initSize := log.Size()
|
||||
|
||||
startSentinel := []byte{0x00, 0x00, 0x00, 0x00}
|
||||
_, err = log.Write(startSentinel)
|
||||
require.NoError(t, err)
|
||||
|
||||
nWritesToFill := 4
|
||||
writeSize := maxCircularLogSize / nWritesToFill
|
||||
for i := 0; i < (nWritesToFill + 1); i++ {
|
||||
_, err := writeFixedSize(log, writeSize)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
endSentinel := []byte{0xFF, 0xFF, 0xFF, 0xFF}
|
||||
_, err = log.Write(endSentinel)
|
||||
require.NoError(t, err)
|
||||
|
||||
finalSize := log.Size()
|
||||
require.Greater(t, finalSize, initSize)
|
||||
require.Greater(t, log.TotalWritten(), finalSize)
|
||||
require.Equal(t, finalSize, maxCircularLogSize)
|
||||
require.Equal(t, finalSize, log.Len())
|
||||
|
||||
logBytes := log.Bytes()
|
||||
|
||||
require.Equal(t, endSentinel, logBytes[len(logBytes)-len(endSentinel):])
|
||||
require.NotEqual(t, startSentinel, logBytes[:len(startSentinel)])
|
||||
})
|
||||
t.Run("overflow-write", func(t *testing.T) {
|
||||
log, err := circlog.NewCircularLog(maxCircularLogSize)
|
||||
require.NoError(t, err)
|
||||
|
||||
initSize := log.Size()
|
||||
writeSize := maxCircularLogSize + 1
|
||||
n, err := writeFixedSize(log, writeSize)
|
||||
require.NoError(t, err)
|
||||
finalSize := log.Size()
|
||||
require.Less(t, n, writeSize)
|
||||
require.Greater(t, finalSize, initSize)
|
||||
require.Equal(t, finalSize, maxCircularLogSize)
|
||||
logBytes := log.Bytes()
|
||||
require.Equal(t, []byte("(...)"), logBytes[:5])
|
||||
})
|
||||
t.Run("stringify", func(t *testing.T) {
|
||||
log, err := circlog.NewCircularLog(maxCircularLogSize)
|
||||
require.NoError(t, err)
|
||||
|
||||
writtenString := "A harmful truth is better than a useful lie."
|
||||
n, err := log.Write([]byte(writtenString))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, len(writtenString), n)
|
||||
loggedString := log.String()
|
||||
require.Equal(t, writtenString, loggedString)
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package circlog
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestNextPow2Int(t *testing.T) {
|
||||
require.Equal(t, 512, nextPow2Int(512), "a power of 2 should round to itself")
|
||||
require.Equal(t, 1024, nextPow2Int(513), "should round up to the next power of 2")
|
||||
require.PanicsWithValue(t, "can only round up positive integers", func() { nextPow2Int(0) }, "unimplemented: zero is not positive; corner case")
|
||||
require.PanicsWithValue(t, "can only round up positive integers", func() { nextPow2Int(-1) }, "unimplemented: cannot round up negative numbers")
|
||||
maxInt := int((^uint(0)) >> 1)
|
||||
require.PanicsWithValue(t, "rounded to larger than int()", func() { nextPow2Int(maxInt - 1) }, "cannot round to a number bigger than the int type")
|
||||
}
|
||||
Reference in New Issue
Block a user