badguardhome/internal/querylog/querylog_search.go
Ainar Garipov 13f708b483 Pull request: * querylog: fix end of log handling
Merge in DNS/adguard-home from 2229-query-log to master

Closes #2229.

Squashed commit of the following:

commit 32508a3f3b1e098869e1649a2774f1f17d14d41f
Author: Ainar Garipov <A.Garipov@AdGuard.COM>
Date:   Tue Nov 10 16:51:25 2020 +0300

    * querylog: add test

commit 774159cc313a0284a8bb8327489671e5d7a3e4eb
Author: Ainar Garipov <A.Garipov@AdGuard.COM>
Date:   Tue Nov 10 15:26:26 2020 +0300

    * querylog: better errors

commit 27b13a4dcaff9e8f9b08aec81c0c03f62ebd3fa5
Author: Ainar Garipov <A.Garipov@AdGuard.COM>
Date:   Tue Nov 10 15:18:51 2020 +0300

    * querylog: fix end of log handling
2020-11-10 19:00:55 +03:00

185 lines
4.7 KiB
Go

package querylog
import (
"errors"
"io"
"time"
"github.com/AdguardTeam/AdGuardHome/internal/util"
"github.com/AdguardTeam/golibs/log"
)
// search - searches log entries in the query log using specified parameters
// returns the list of entries found + time of the oldest entry
func (l *queryLog) search(params *searchParams) ([]*logEntry, time.Time) {
now := time.Now()
if params.limit == 0 {
return []*logEntry{}, time.Time{}
}
// add from file
fileEntries, oldest, total := l.searchFiles(params)
// add from memory buffer
l.bufferLock.Lock()
total += len(l.buffer)
memoryEntries := make([]*logEntry, 0)
// go through the buffer in the reverse order
// from NEWER to OLDER
for i := len(l.buffer) - 1; i >= 0; i-- {
entry := l.buffer[i]
if !params.match(entry) {
continue
}
memoryEntries = append(memoryEntries, entry)
}
l.bufferLock.Unlock()
// limits
totalLimit := params.offset + params.limit
// now let's get a unified collection
entries := append(memoryEntries, fileEntries...)
if len(entries) > totalLimit {
// remove extra records
entries = entries[:totalLimit]
}
if params.offset > 0 {
if len(entries) > params.offset {
entries = entries[params.offset:]
} else {
entries = make([]*logEntry, 0)
oldest = time.Time{}
}
}
if len(entries) > 0 && len(entries) <= totalLimit {
// Update oldest after merging in the memory buffer.
oldest = entries[len(entries)-1].Time
}
log.Debug("QueryLog: prepared data (%d/%d) older than %s in %s",
len(entries), total, params.olderThan, time.Since(now))
return entries, oldest
}
// searchFiles reads log entries from all log files and applies the specified search criteria.
// IMPORTANT: this method does not scan more than "maxSearchEntries" so you
// may need to call it many times.
//
// it returns:
// * an array of log entries that we have read
// * time of the oldest processed entry (even if it was discarded)
// * total number of processed entries (including discarded).
func (l *queryLog) searchFiles(params *searchParams) ([]*logEntry, time.Time, int) {
entries := make([]*logEntry, 0)
oldest := time.Time{}
r, err := l.openReader()
if err != nil {
log.Error("Failed to open qlog reader: %v", err)
return entries, oldest, 0
}
defer r.Close()
if params.olderThan.IsZero() {
err = r.SeekStart()
} else {
err = r.Seek(params.olderThan.UnixNano())
if err == nil {
// Read to the next record right away
// The one that was specified in the "oldest" param is not needed,
// we need only the one next to it
_, err = r.ReadNext()
} else if errors.Is(err, ErrEndOfLog) {
// We've reached the end of the log.
return entries, time.Time{}, 0
}
}
if err != nil {
log.Debug("Cannot Seek() to %v: %v", params.olderThan, err)
return entries, oldest, 0
}
totalLimit := params.offset + params.limit
total := 0
oldestNano := int64(0)
// By default, we do not scan more than "maxFileScanEntries" at once
// The idea is to make search calls faster so that the UI could handle it and show something
// This behavior can be overridden if "maxFileScanEntries" is set to 0
for total < params.maxFileScanEntries || params.maxFileScanEntries <= 0 {
entry, ts, err := l.readNextEntry(r, params)
if err == io.EOF {
// there's nothing to read anymore
break
}
oldestNano = ts
total++
if entry != nil {
entries = append(entries, entry)
if len(entries) == totalLimit {
// Do not read more than "totalLimit" records at once
break
}
}
}
if oldestNano != 0 {
oldest = time.Unix(0, oldestNano)
}
return entries, oldest, total
}
// readNextEntry - reads the next log entry and checks if it matches the search criteria (getDataParams)
//
// returns:
// * log entry that matches search criteria or null if it was discarded (or if there's nothing to read)
// * timestamp of the processed log entry
// * error if we can't read anymore
func (l *queryLog) readNextEntry(r *QLogReader, params *searchParams) (*logEntry, int64, error) {
line, err := r.ReadNext()
if err != nil {
return nil, 0, err
}
// Read the log record timestamp right away
timestamp := readQLogTimestamp(line)
// Quick check without deserializing log entry
if !params.quickMatch(line) {
return nil, timestamp, nil
}
entry := logEntry{}
decodeLogEntry(&entry, line)
// Full check of the deserialized log entry
if !params.match(&entry) {
return nil, timestamp, nil
}
return &entry, timestamp, nil
}
// openReader - opens QLogReader instance
func (l *queryLog) openReader() (*QLogReader, error) {
files := make([]string, 0)
if util.FileExists(l.logFile + ".1") {
files = append(files, l.logFile+".1")
}
if util.FileExists(l.logFile) {
files = append(files, l.logFile)
}
return NewQLogReader(files)
}