all: sync with master; upd chlog

internal/querylog/qlog.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/AdguardTeam/AdGuardHome/internal/aghalg"
 	"github.com/AdguardTeam/AdGuardHome/internal/aghnet"
 	"github.com/AdguardTeam/AdGuardHome/internal/filtering"
 	"github.com/AdguardTeam/golibs/errors"
@@ -29,12 +30,12 @@ type queryLog struct {
 
 	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
 	// be modified.
-	buffer []*logEntry
+	buffer *aghalg.RingBuffer[*logEntry]
+
+	// logFile is the path to the log file.
+	logFile string
 
 	// bufferLock protects buffer.
 	bufferLock sync.RWMutex
@@ -195,7 +196,7 @@ func newLogEntry(params *AddParams) (entry *logEntry) {
 // Add implements the [QueryLog] interface for *queryLog.
 func (l *queryLog) Add(params *AddParams) {
 	var isEnabled, fileIsEnabled bool
-	var memSize uint32
+	var memSize int
 	func() {
 		l.confMu.RLock()
 		defer l.confMu.RUnlock()
@@ -204,7 +205,7 @@ func (l *queryLog) Add(params *AddParams) {
 		memSize = l.conf.MemSize
 	}()
 
-	if !isEnabled {
+	if !isEnabled || memSize == 0 {
 		return
 	}
 
@@ -221,36 +222,18 @@ func (l *queryLog) Add(params *AddParams) {
 
 	entry := newLogEntry(params)
 
-	needFlush := false
-	func() {
-		l.bufferLock.Lock()
-		defer l.bufferLock.Unlock()
+	l.bufferLock.Lock()
+	defer l.bufferLock.Unlock()
 
-		l.buffer = append(l.buffer, entry)
+	l.buffer.Append(entry)
 
-		if !fileIsEnabled {
-			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
-			}
-		}
-	}()
+	if !l.flushPending && fileIsEnabled && l.buffer.Len() >= memSize {
+		l.flushPending = true
 
-	if needFlush {
 		go func() {
 			flushErr := l.flushLogBuffer()
 			if flushErr != nil {
-				log.Error("querylog: flushing after adding: %s", err)
+				log.Error("querylog: flushing after adding: %s", flushErr)
 			}
 		}()
 	}

internal/querylog/querylog.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/AdguardTeam/AdGuardHome/internal/aghalg"
 	"github.com/AdguardTeam/AdGuardHome/internal/aghhttp"
 	"github.com/AdguardTeam/AdGuardHome/internal/aghnet"
 	"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
 	// flushed to disk.
-	MemSize uint32
+	MemSize int
 
 	// Enabled tells if the query log is enabled.
 	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{
 		findClient: findClient,
 
+		buffer: aghalg.NewRingBuffer[*logEntry](conf.MemSize),
+
 		conf:    &Config{},
 		confMu:  &sync.RWMutex{},
 		logFile: filepath.Join(conf.BaseDir, queryLogFileName),

internal/querylog/querylogfile.go

@@ -14,68 +14,75 @@ import (
 // flushLogBuffer flushes the current buffer to file and resets the current
 // buffer.
 func (l *queryLog) flushLogBuffer() (err error) {
+	defer func() { err = errors.Annotate(err, "flushing log buffer: %w") }()
 	l.fileFlushLock.Lock()
 	defer l.fileFlushLock.Unlock()
 
-	var flushBuffer []*logEntry
-	func() {
-		l.bufferLock.Lock()
-		defer l.bufferLock.Unlock()
+	b, err := l.encodeEntries()
+	if err != nil {
+		// Don't wrap the error since it's informative enough as is.
+		return err
+	}
 
-		flushBuffer = l.buffer
-		l.buffer = nil
-		l.flushPending = false
-	}()
-
-	err = l.flushToFile(flushBuffer)
-
-	return errors.Annotate(err, "writing to file: %w")
+	return l.flushToFile(b)
 }
 
-// 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")
+// 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()
+	defer l.bufferLock.Unlock()
 
-		return nil
+	bufLen := l.buffer.Len()
+	if bufLen == 0 {
+		return nil, errors.Error("nothing to write to a file")
 	}
 
 	start := time.Now()
 
-	var b bytes.Buffer
-	e := json.NewEncoder(&b)
-	for _, entry := range buffer {
-		err = e.Encode(entry)
-		if err != nil {
-			log.Error("Failed to marshal entry: %s", err)
+	b = &bytes.Buffer{}
+	e := json.NewEncoder(b)
 
-			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)
-	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
-	filename := l.logFile
-	zb = b
+	l.buffer.Clear()
+	l.flushPending = false
 
+	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()
 	defer l.fileWriteLock.Unlock()
+
+	filename := l.logFile
+
 	f, err := os.OpenFile(filename, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o644)
 	if err != nil {
-		log.Error("failed to create file \"%s\": %s", filename, err)
-		return err
+		return fmt.Errorf("creating file %q: %w", filename, err)
 	}
+
 	defer func() { err = errors.WithDeferred(err, f.Close()) }()
 
-	n, err := f.Write(zb.Bytes())
+	n, err := f.Write(b.Bytes())
 	if err != nil {
-		log.Error("Couldn't write to file: %s", err)
-		return err
+		return fmt.Errorf("writing to file %q: %w", filename, err)
 	}
 
-	log.Debug("querylog: ok \"%s\": %v bytes written", filename, n)
+	log.Debug("querylog: ok %q: %v bytes written", filename, n)
 
 	return nil
 }

internal/querylog/search.go

@@ -51,13 +51,12 @@ func (l *queryLog) searchMemory(params *searchParams, cache clientCache) (entrie
 	l.bufferLock.Lock()
 	defer l.bufferLock.Unlock()
 
-	// Go through the buffer in the reverse order, from newer to older.
-	var err error
-	for i := len(l.buffer) - 1; i >= 0; i-- {
+	l.buffer.ReverseRange(func(entry *logEntry) (cont bool) {
 		// A shallow clone is enough, since the only thing that this loop
 		// 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)
 		if err != nil {
 			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) {
 			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