Pull request 2020: AG-26236-ring-buffer
Squashed commit of the following:
commit 4b9cc9ddf52739fc5f918babedc99ac7ac0e2415
Merge: a6259ed57 39aeaf891
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Tue Oct 3 20:39:58 2023 +0300
Merge branch 'master' into AG-26236-ring-buffer
commit a6259ed5758156e4110ee3ea6a49760d2880ade3
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Tue Oct 3 20:30:20 2023 +0300
querylog: imp code
commit 40f9f7cd5a1cff22bcb858020f2cfa9be8399671
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Tue Oct 3 20:11:49 2023 +0300
querylog: fix typo
commit 1aabbadcb5fcbe6a95945c5bd1455b956b85a8d0
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Tue Oct 3 20:05:22 2023 +0300
querylog: imp err msg
commit 02913d35b43e190e42765823ccfcdd332839e984
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Tue Oct 3 19:37:47 2023 +0300
aghalg: imp tests
commit 98a7909088d2a65b78afa9fa3113545c94429e65
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Tue Oct 3 19:23:10 2023 +0300
all: imp tests
commit e147804eeafe89e5020cd917784eef3ff2b310d0
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Tue Oct 3 18:49:49 2023 +0300
all: imp code
commit 5f21f2f63b7bce89f2bc79c97c7350cd693c956b
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Tue Oct 3 14:39:43 2023 +0300
all: add tests
commit 35a45c7dc5d5961f6987da7c69249c56e54e97f5
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Mon Oct 2 17:43:09 2023 +0300
all: imp code
commit 21e51fcbe411258eaf830825df9d05b7ddcc187a
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Fri Sep 29 18:21:00 2023 +0300
all: add ring buffer
This commit is contained in:
parent
39aeaf8910
commit
4479b32ad4
|
@ -0,0 +1,102 @@
|
||||||
|
package aghalg
|
||||||
|
|
||||||
|
import (
|
||||||
|
"github.com/AdguardTeam/golibs/errors"
|
||||||
|
)
|
||||||
|
|
||||||
|
// RingBuffer is the implementation of ring buffer data structure.
|
||||||
|
type RingBuffer[T any] struct {
|
||||||
|
buf []T
|
||||||
|
cur int
|
||||||
|
full bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewRingBuffer initializes the new instance of ring buffer. size must be
|
||||||
|
// greater or equal to zero.
|
||||||
|
func NewRingBuffer[T any](size int) (rb *RingBuffer[T]) {
|
||||||
|
if size < 0 {
|
||||||
|
panic(errors.Error("ring buffer: size must be greater or equal to zero"))
|
||||||
|
}
|
||||||
|
|
||||||
|
return &RingBuffer[T]{
|
||||||
|
buf: make([]T, size),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Append appends an element to the buffer.
|
||||||
|
func (rb *RingBuffer[T]) Append(e T) {
|
||||||
|
if len(rb.buf) == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
rb.buf[rb.cur] = e
|
||||||
|
rb.cur = (rb.cur + 1) % cap(rb.buf)
|
||||||
|
if rb.cur == 0 {
|
||||||
|
rb.full = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Range calls cb for each element of the buffer. If cb returns false it stops.
|
||||||
|
func (rb *RingBuffer[T]) Range(cb func(T) (cont bool)) {
|
||||||
|
before, after := rb.splitCur()
|
||||||
|
|
||||||
|
for _, e := range before {
|
||||||
|
if !cb(e) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, e := range after {
|
||||||
|
if !cb(e) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ReverseRange calls cb for each element of the buffer in reverse order. If
|
||||||
|
// cb returns false it stops.
|
||||||
|
func (rb *RingBuffer[T]) ReverseRange(cb func(T) (cont bool)) {
|
||||||
|
before, after := rb.splitCur()
|
||||||
|
|
||||||
|
for i := len(after) - 1; i >= 0; i-- {
|
||||||
|
if !cb(after[i]) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for i := len(before) - 1; i >= 0; i-- {
|
||||||
|
if !cb(before[i]) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// splitCur splits the buffer in two, before and after current position in
|
||||||
|
// chronological order. If buffer is not full, after is nil.
|
||||||
|
func (rb *RingBuffer[T]) splitCur() (before, after []T) {
|
||||||
|
if len(rb.buf) == 0 {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
cur := rb.cur
|
||||||
|
if !rb.full {
|
||||||
|
return rb.buf[:cur], nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return rb.buf[cur:], rb.buf[:cur]
|
||||||
|
}
|
||||||
|
|
||||||
|
// Len returns a length of the buffer.
|
||||||
|
func (rb *RingBuffer[T]) Len() (l int) {
|
||||||
|
if !rb.full {
|
||||||
|
return rb.cur
|
||||||
|
}
|
||||||
|
|
||||||
|
return cap(rb.buf)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Clear clears the buffer.
|
||||||
|
func (rb *RingBuffer[T]) Clear() {
|
||||||
|
rb.full = false
|
||||||
|
rb.cur = 0
|
||||||
|
}
|
|
@ -0,0 +1,173 @@
|
||||||
|
package aghalg_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/AdguardTeam/AdGuardHome/internal/aghalg"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"golang.org/x/exp/slices"
|
||||||
|
)
|
||||||
|
|
||||||
|
// elements is a helper function that returns n elements of the buffer.
|
||||||
|
func elements(b *aghalg.RingBuffer[int], n int, reverse bool) (es []int) {
|
||||||
|
fn := b.Range
|
||||||
|
if reverse {
|
||||||
|
fn = b.ReverseRange
|
||||||
|
}
|
||||||
|
|
||||||
|
i := 0
|
||||||
|
fn(func(e int) (cont bool) {
|
||||||
|
if i >= n {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
es = append(es, e)
|
||||||
|
i++
|
||||||
|
|
||||||
|
return true
|
||||||
|
})
|
||||||
|
|
||||||
|
return es
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNewRingBuffer(t *testing.T) {
|
||||||
|
t.Run("success_and_clear", func(t *testing.T) {
|
||||||
|
b := aghalg.NewRingBuffer[int](5)
|
||||||
|
for i := 0; i < 10; i++ {
|
||||||
|
b.Append(i)
|
||||||
|
}
|
||||||
|
assert.Equal(t, []int{5, 6, 7, 8, 9}, elements(b, b.Len(), false))
|
||||||
|
|
||||||
|
b.Clear()
|
||||||
|
assert.Zero(t, b.Len())
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("negative_size", func(t *testing.T) {
|
||||||
|
assert.PanicsWithError(t, "ring buffer: size must be greater or equal to zero", func() {
|
||||||
|
aghalg.NewRingBuffer[int](-5)
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("zero", func(t *testing.T) {
|
||||||
|
b := aghalg.NewRingBuffer[int](0)
|
||||||
|
for i := 0; i < 10; i++ {
|
||||||
|
b.Append(i)
|
||||||
|
assert.Equal(t, 0, b.Len())
|
||||||
|
assert.Empty(t, elements(b, b.Len(), false))
|
||||||
|
assert.Empty(t, elements(b, b.Len(), true))
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("single", func(t *testing.T) {
|
||||||
|
b := aghalg.NewRingBuffer[int](1)
|
||||||
|
for i := 0; i < 10; i++ {
|
||||||
|
b.Append(i)
|
||||||
|
assert.Equal(t, 1, b.Len())
|
||||||
|
assert.Equal(t, []int{i}, elements(b, b.Len(), false))
|
||||||
|
assert.Equal(t, []int{i}, elements(b, b.Len(), true))
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRingBuffer_Range(t *testing.T) {
|
||||||
|
const size = 5
|
||||||
|
|
||||||
|
b := aghalg.NewRingBuffer[int](size)
|
||||||
|
|
||||||
|
testCases := []struct {
|
||||||
|
name string
|
||||||
|
want []int
|
||||||
|
count int
|
||||||
|
length int
|
||||||
|
}{{
|
||||||
|
name: "three",
|
||||||
|
count: 3,
|
||||||
|
length: 3,
|
||||||
|
want: []int{0, 1, 2},
|
||||||
|
}, {
|
||||||
|
name: "ten",
|
||||||
|
count: 10,
|
||||||
|
length: size,
|
||||||
|
want: []int{5, 6, 7, 8, 9},
|
||||||
|
}, {
|
||||||
|
name: "hundred",
|
||||||
|
count: 100,
|
||||||
|
length: size,
|
||||||
|
want: []int{95, 96, 97, 98, 99},
|
||||||
|
}}
|
||||||
|
|
||||||
|
for _, tc := range testCases {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
for i := 0; i < tc.count; i++ {
|
||||||
|
b.Append(i)
|
||||||
|
}
|
||||||
|
|
||||||
|
bufLen := b.Len()
|
||||||
|
assert.Equal(t, tc.length, bufLen)
|
||||||
|
|
||||||
|
want := tc.want
|
||||||
|
assert.Equal(t, want, elements(b, bufLen, false))
|
||||||
|
assert.Equal(t, want[:len(want)-1], elements(b, bufLen-1, false))
|
||||||
|
assert.Equal(t, want[:len(want)/2], elements(b, bufLen/2, false))
|
||||||
|
|
||||||
|
want = want[:cap(want)]
|
||||||
|
slices.Reverse(want)
|
||||||
|
|
||||||
|
assert.Equal(t, want, elements(b, bufLen, true))
|
||||||
|
assert.Equal(t, want[:len(want)-1], elements(b, bufLen-1, true))
|
||||||
|
assert.Equal(t, want[:len(want)/2], elements(b, bufLen/2, true))
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRingBuffer_Range_increment(t *testing.T) {
|
||||||
|
const size = 5
|
||||||
|
|
||||||
|
b := aghalg.NewRingBuffer[int](size)
|
||||||
|
|
||||||
|
testCases := []struct {
|
||||||
|
name string
|
||||||
|
want []int
|
||||||
|
}{{
|
||||||
|
name: "one",
|
||||||
|
want: []int{0},
|
||||||
|
}, {
|
||||||
|
name: "two",
|
||||||
|
want: []int{0, 1},
|
||||||
|
}, {
|
||||||
|
name: "three",
|
||||||
|
want: []int{0, 1, 2},
|
||||||
|
}, {
|
||||||
|
name: "four",
|
||||||
|
want: []int{0, 1, 2, 3},
|
||||||
|
}, {
|
||||||
|
name: "five",
|
||||||
|
want: []int{0, 1, 2, 3, 4},
|
||||||
|
}, {
|
||||||
|
name: "six",
|
||||||
|
want: []int{1, 2, 3, 4, 5},
|
||||||
|
}, {
|
||||||
|
name: "seven",
|
||||||
|
want: []int{2, 3, 4, 5, 6},
|
||||||
|
}, {
|
||||||
|
name: "eight",
|
||||||
|
want: []int{3, 4, 5, 6, 7},
|
||||||
|
}, {
|
||||||
|
name: "nine",
|
||||||
|
want: []int{4, 5, 6, 7, 8},
|
||||||
|
}, {
|
||||||
|
name: "ten",
|
||||||
|
want: []int{5, 6, 7, 8, 9},
|
||||||
|
}}
|
||||||
|
|
||||||
|
for i, tc := range testCases {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
b.Append(i)
|
||||||
|
|
||||||
|
assert.Equal(t, tc.want, elements(b, b.Len(), false))
|
||||||
|
|
||||||
|
slices.Reverse(tc.want)
|
||||||
|
assert.Equal(t, tc.want, elements(b, b.Len(), true))
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
|
@ -262,7 +262,7 @@ type queryLogConfig struct {
|
||||||
|
|
||||||
// MemSize is the number of entries kept in memory before they are flushed
|
// MemSize is the number of entries kept in memory before they are flushed
|
||||||
// to disk.
|
// to disk.
|
||||||
MemSize uint32 `yaml:"size_memory"`
|
MemSize int `yaml:"size_memory"`
|
||||||
|
|
||||||
// Enabled defines if the query log is enabled.
|
// Enabled defines if the query log is enabled.
|
||||||
Enabled bool `yaml:"enabled"`
|
Enabled bool `yaml:"enabled"`
|
||||||
|
|
|
@ -7,6 +7,7 @@ import (
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/AdguardTeam/AdGuardHome/internal/aghalg"
|
||||||
"github.com/AdguardTeam/AdGuardHome/internal/aghnet"
|
"github.com/AdguardTeam/AdGuardHome/internal/aghnet"
|
||||||
"github.com/AdguardTeam/AdGuardHome/internal/filtering"
|
"github.com/AdguardTeam/AdGuardHome/internal/filtering"
|
||||||
"github.com/AdguardTeam/golibs/errors"
|
"github.com/AdguardTeam/golibs/errors"
|
||||||
|
@ -29,12 +30,12 @@ type queryLog struct {
|
||||||
|
|
||||||
findClient func(ids []string) (c *Client, err error)
|
findClient func(ids []string) (c *Client, err error)
|
||||||
|
|
||||||
// logFile is the path to the log file.
|
|
||||||
logFile string
|
|
||||||
|
|
||||||
// buffer contains recent log entries. The entries in this buffer must not
|
// buffer contains recent log entries. The entries in this buffer must not
|
||||||
// be modified.
|
// be modified.
|
||||||
buffer []*logEntry
|
buffer *aghalg.RingBuffer[*logEntry]
|
||||||
|
|
||||||
|
// logFile is the path to the log file.
|
||||||
|
logFile string
|
||||||
|
|
||||||
// bufferLock protects buffer.
|
// bufferLock protects buffer.
|
||||||
bufferLock sync.RWMutex
|
bufferLock sync.RWMutex
|
||||||
|
@ -195,7 +196,7 @@ func newLogEntry(params *AddParams) (entry *logEntry) {
|
||||||
// Add implements the [QueryLog] interface for *queryLog.
|
// Add implements the [QueryLog] interface for *queryLog.
|
||||||
func (l *queryLog) Add(params *AddParams) {
|
func (l *queryLog) Add(params *AddParams) {
|
||||||
var isEnabled, fileIsEnabled bool
|
var isEnabled, fileIsEnabled bool
|
||||||
var memSize uint32
|
var memSize int
|
||||||
func() {
|
func() {
|
||||||
l.confMu.RLock()
|
l.confMu.RLock()
|
||||||
defer l.confMu.RUnlock()
|
defer l.confMu.RUnlock()
|
||||||
|
@ -204,7 +205,7 @@ func (l *queryLog) Add(params *AddParams) {
|
||||||
memSize = l.conf.MemSize
|
memSize = l.conf.MemSize
|
||||||
}()
|
}()
|
||||||
|
|
||||||
if !isEnabled {
|
if !isEnabled || memSize == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -221,36 +222,18 @@ func (l *queryLog) Add(params *AddParams) {
|
||||||
|
|
||||||
entry := newLogEntry(params)
|
entry := newLogEntry(params)
|
||||||
|
|
||||||
needFlush := false
|
|
||||||
func() {
|
|
||||||
l.bufferLock.Lock()
|
l.bufferLock.Lock()
|
||||||
defer l.bufferLock.Unlock()
|
defer l.bufferLock.Unlock()
|
||||||
|
|
||||||
l.buffer = append(l.buffer, entry)
|
l.buffer.Append(entry)
|
||||||
|
|
||||||
if !fileIsEnabled {
|
if !l.flushPending && fileIsEnabled && l.buffer.Len() >= memSize {
|
||||||
if len(l.buffer) > int(memSize) {
|
|
||||||
// Writing to file is disabled, so just remove the oldest entry
|
|
||||||
// from the slices.
|
|
||||||
//
|
|
||||||
// TODO(a.garipov): This should be replaced by a proper ring
|
|
||||||
// buffer, but it's currently difficult to do that.
|
|
||||||
l.buffer[0] = nil
|
|
||||||
l.buffer = l.buffer[1:]
|
|
||||||
}
|
|
||||||
} else if !l.flushPending {
|
|
||||||
needFlush = len(l.buffer) >= int(memSize)
|
|
||||||
if needFlush {
|
|
||||||
l.flushPending = true
|
l.flushPending = true
|
||||||
}
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
if needFlush {
|
|
||||||
go func() {
|
go func() {
|
||||||
flushErr := l.flushLogBuffer()
|
flushErr := l.flushLogBuffer()
|
||||||
if flushErr != nil {
|
if flushErr != nil {
|
||||||
log.Error("querylog: flushing after adding: %s", err)
|
log.Error("querylog: flushing after adding: %s", flushErr)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|
|
@ -7,6 +7,7 @@ import (
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/AdguardTeam/AdGuardHome/internal/aghalg"
|
||||||
"github.com/AdguardTeam/AdGuardHome/internal/aghhttp"
|
"github.com/AdguardTeam/AdGuardHome/internal/aghhttp"
|
||||||
"github.com/AdguardTeam/AdGuardHome/internal/aghnet"
|
"github.com/AdguardTeam/AdGuardHome/internal/aghnet"
|
||||||
"github.com/AdguardTeam/AdGuardHome/internal/filtering"
|
"github.com/AdguardTeam/AdGuardHome/internal/filtering"
|
||||||
|
@ -62,7 +63,7 @@ type Config struct {
|
||||||
|
|
||||||
// MemSize is the number of entries kept in a memory buffer before they are
|
// MemSize is the number of entries kept in a memory buffer before they are
|
||||||
// flushed to disk.
|
// flushed to disk.
|
||||||
MemSize uint32
|
MemSize int
|
||||||
|
|
||||||
// Enabled tells if the query log is enabled.
|
// Enabled tells if the query log is enabled.
|
||||||
Enabled bool
|
Enabled bool
|
||||||
|
@ -142,9 +143,15 @@ func newQueryLog(conf Config) (l *queryLog, err error) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if conf.MemSize < 0 {
|
||||||
|
return nil, errors.Error("memory size must be greater or equal to zero")
|
||||||
|
}
|
||||||
|
|
||||||
l = &queryLog{
|
l = &queryLog{
|
||||||
findClient: findClient,
|
findClient: findClient,
|
||||||
|
|
||||||
|
buffer: aghalg.NewRingBuffer[*logEntry](conf.MemSize),
|
||||||
|
|
||||||
conf: &Config{},
|
conf: &Config{},
|
||||||
confMu: &sync.RWMutex{},
|
confMu: &sync.RWMutex{},
|
||||||
logFile: filepath.Join(conf.BaseDir, queryLogFileName),
|
logFile: filepath.Join(conf.BaseDir, queryLogFileName),
|
||||||
|
|
|
@ -14,68 +14,75 @@ import (
|
||||||
// flushLogBuffer flushes the current buffer to file and resets the current
|
// flushLogBuffer flushes the current buffer to file and resets the current
|
||||||
// buffer.
|
// buffer.
|
||||||
func (l *queryLog) flushLogBuffer() (err error) {
|
func (l *queryLog) flushLogBuffer() (err error) {
|
||||||
|
defer func() { err = errors.Annotate(err, "flushing log buffer: %w") }()
|
||||||
l.fileFlushLock.Lock()
|
l.fileFlushLock.Lock()
|
||||||
defer l.fileFlushLock.Unlock()
|
defer l.fileFlushLock.Unlock()
|
||||||
|
|
||||||
var flushBuffer []*logEntry
|
b, err := l.encodeEntries()
|
||||||
func() {
|
if err != nil {
|
||||||
|
// Don't wrap the error since it's informative enough as is.
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
return l.flushToFile(b)
|
||||||
|
}
|
||||||
|
|
||||||
|
// encodeEntries returns JSON encoded log entries, logs estimated time, clears
|
||||||
|
// the log buffer.
|
||||||
|
func (l *queryLog) encodeEntries() (b *bytes.Buffer, err error) {
|
||||||
l.bufferLock.Lock()
|
l.bufferLock.Lock()
|
||||||
defer l.bufferLock.Unlock()
|
defer l.bufferLock.Unlock()
|
||||||
|
|
||||||
flushBuffer = l.buffer
|
bufLen := l.buffer.Len()
|
||||||
l.buffer = nil
|
if bufLen == 0 {
|
||||||
l.flushPending = false
|
return nil, errors.Error("nothing to write to a file")
|
||||||
}()
|
|
||||||
|
|
||||||
err = l.flushToFile(flushBuffer)
|
|
||||||
|
|
||||||
return errors.Annotate(err, "writing to file: %w")
|
|
||||||
}
|
|
||||||
|
|
||||||
// flushToFile saves the specified log entries to the query log file
|
|
||||||
func (l *queryLog) flushToFile(buffer []*logEntry) (err error) {
|
|
||||||
if len(buffer) == 0 {
|
|
||||||
log.Debug("querylog: nothing to write to a file")
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
|
|
||||||
var b bytes.Buffer
|
b = &bytes.Buffer{}
|
||||||
e := json.NewEncoder(&b)
|
e := json.NewEncoder(b)
|
||||||
for _, entry := range buffer {
|
|
||||||
err = e.Encode(entry)
|
|
||||||
if err != nil {
|
|
||||||
log.Error("Failed to marshal entry: %s", err)
|
|
||||||
|
|
||||||
return err
|
l.buffer.Range(func(entry *logEntry) (cont bool) {
|
||||||
}
|
err = e.Encode(entry)
|
||||||
|
|
||||||
|
return err == nil
|
||||||
|
})
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
// Don't wrap the error since it's informative enough as is.
|
||||||
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
elapsed := time.Since(start)
|
elapsed := time.Since(start)
|
||||||
log.Debug("%d elements serialized via json in %v: %d kB, %v/entry, %v/entry", len(buffer), elapsed, b.Len()/1024, float64(b.Len())/float64(len(buffer)), elapsed/time.Duration(len(buffer)))
|
log.Debug("%d elements serialized via json in %v: %d kB, %v/entry, %v/entry", bufLen, elapsed, b.Len()/1024, float64(b.Len())/float64(bufLen), elapsed/time.Duration(bufLen))
|
||||||
|
|
||||||
var zb bytes.Buffer
|
l.buffer.Clear()
|
||||||
filename := l.logFile
|
l.flushPending = false
|
||||||
zb = b
|
|
||||||
|
|
||||||
|
return b, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// flushToFile saves the encoded log entries to the query log file.
|
||||||
|
func (l *queryLog) flushToFile(b *bytes.Buffer) (err error) {
|
||||||
l.fileWriteLock.Lock()
|
l.fileWriteLock.Lock()
|
||||||
defer l.fileWriteLock.Unlock()
|
defer l.fileWriteLock.Unlock()
|
||||||
|
|
||||||
|
filename := l.logFile
|
||||||
|
|
||||||
f, err := os.OpenFile(filename, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o644)
|
f, err := os.OpenFile(filename, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o644)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Error("failed to create file \"%s\": %s", filename, err)
|
return fmt.Errorf("creating file %q: %w", filename, err)
|
||||||
return err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() { err = errors.WithDeferred(err, f.Close()) }()
|
defer func() { err = errors.WithDeferred(err, f.Close()) }()
|
||||||
|
|
||||||
n, err := f.Write(zb.Bytes())
|
n, err := f.Write(b.Bytes())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Error("Couldn't write to file: %s", err)
|
return fmt.Errorf("writing to file %q: %w", filename, err)
|
||||||
return err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
log.Debug("querylog: ok \"%s\": %v bytes written", filename, n)
|
log.Debug("querylog: ok %q: %v bytes written", filename, n)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
|
@ -51,13 +51,12 @@ func (l *queryLog) searchMemory(params *searchParams, cache clientCache) (entrie
|
||||||
l.bufferLock.Lock()
|
l.bufferLock.Lock()
|
||||||
defer l.bufferLock.Unlock()
|
defer l.bufferLock.Unlock()
|
||||||
|
|
||||||
// Go through the buffer in the reverse order, from newer to older.
|
l.buffer.ReverseRange(func(entry *logEntry) (cont bool) {
|
||||||
var err error
|
|
||||||
for i := len(l.buffer) - 1; i >= 0; i-- {
|
|
||||||
// A shallow clone is enough, since the only thing that this loop
|
// A shallow clone is enough, since the only thing that this loop
|
||||||
// modifies is the client field.
|
// modifies is the client field.
|
||||||
e := l.buffer[i].shallowClone()
|
e := entry.shallowClone()
|
||||||
|
|
||||||
|
var err error
|
||||||
e.client, err = l.client(e.ClientID, e.IP.String(), cache)
|
e.client, err = l.client(e.ClientID, e.IP.String(), cache)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
msg := "querylog: enriching memory record at time %s" +
|
msg := "querylog: enriching memory record at time %s" +
|
||||||
|
@ -70,9 +69,11 @@ func (l *queryLog) searchMemory(params *searchParams, cache clientCache) (entrie
|
||||||
if params.match(e) {
|
if params.match(e) {
|
||||||
entries = append(entries, e)
|
entries = append(entries, e)
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
return entries, len(l.buffer)
|
return true
|
||||||
|
})
|
||||||
|
|
||||||
|
return entries, l.buffer.Len()
|
||||||
}
|
}
|
||||||
|
|
||||||
// search - searches log entries in the query log using specified parameters
|
// search - searches log entries in the query log using specified parameters
|
||||||
|
|
Loading…
Reference in New Issue