- querylog: file rotation didn't work properly; fix entry searching algorithm
If AGH is restarted, file rotation timer is reset
which can lead to the situation when file rotation procedure is never started.
Squashed commit of the following:
commit 427ae91a512cd146ebfffad06ed24eb723cb9e7d
Merge: 067fac65 e56c746b
Author: Simon Zolin <s.zolin@adguard.com>
Date: Wed Sep 2 18:18:46 2020 +0300
Merge remote-tracking branch 'origin/master' into qlogs-rotate
commit 067fac65b1a87d499900f4860ffa96ed8208967c
Author: Simon Zolin <s.zolin@adguard.com>
Date: Wed Sep 2 15:30:48 2020 +0300
minor
commit c2059a15700e5696cb1bb5cd49129c6020d986f4
Author: Simon Zolin <s.zolin@adguard.com>
Date: Wed Sep 2 14:53:07 2020 +0300
improve
commit a279438eaf1cf40b820652093fb56d56784de7d8
Author: Simon Zolin <s.zolin@adguard.com>
Date: Tue Sep 1 18:49:14 2020 +0300
minor
commit 26ac130f139f565de39200e484b3bd4a04afcfcc
Author: Simon Zolin <s.zolin@adguard.com>
Date: Tue Sep 1 13:54:27 2020 +0300
rename
commit 0fad7b88dbeadcddd4d77536a18da72f3203ea80
Author: Simon Zolin <s.zolin@adguard.com>
Date: Tue Sep 1 13:05:36 2020 +0300
+ TestQLogSeek
commit fa6afc6d4dc592b1fef67c4a069ea50fae600a58
Author: Simon Zolin <s.zolin@adguard.com>
Date: Tue Sep 1 13:05:34 2020 +0300
minor
commit 11e6ab9131e5c37467e8530a2db95a82bbb0603b
Author: Simon Zolin <s.zolin@adguard.com>
Date: Mon Aug 31 19:45:47 2020 +0300
fix tests
commit 7cbb89948df0e69b1bae8f8cde1879b5b1c4b1d6
Author: Simon Zolin <s.zolin@adguard.com>
Date: Mon Aug 31 19:29:43 2020 +0300
- querylog: fix entry searching algorithm
commit 745d44863d88b321bd7001f24a68620f7ef05819
Author: Simon Zolin <s.zolin@adguard.com>
Date: Mon Aug 31 18:34:14 2020 +0300
- querylog: file rotation didn't work properly
If AGH is restarted, file rotation timer is reset
which can lead to the situation when file rotation procedure is never started.
This commit is contained in:
Simon Zolin
parent
e56c746b60
commit
4efc464e98
@ -1439,6 +1439,11 @@ When UI asks for data from query log (see "API: Get query log"), server reads th
|
||||
|
||||
We store data for a limited amount of time - the log file is automatically rotated.
|
||||
|
||||
* On AGH startup read the first line from query logs and store its time value
|
||||
* If there's no log file yet, set the time value of the first log event when the file is created
|
||||
* If this time value is older than our time limit, perform file rotate procedure
|
||||
* While AGH is running, check the previous condition every 24 hours
|
||||
|
||||
|
||||
### API: Get query log
|
||||
|
||||
|
||||
@ -86,8 +86,10 @@ func (q *QLogFile) Seek(timestamp int64) (int64, int, error) {
|
||||
end := fileInfo.Size() // end of the search interval (position in the file)
|
||||
probe := (end - start) / 2 // probe -- approximate index of the line we'll try to check
|
||||
var line string
|
||||
var lineIdx int64 // index of the probe line in the file
|
||||
var lineIdx int64 // index of the probe line in the file
|
||||
var lineEndIdx int64
|
||||
var lastProbeLineIdx int64 // index of the last probe line
|
||||
lastProbeLineIdx = -1
|
||||
|
||||
// Count seek depth in order to detect mistakes
|
||||
// If depth is too large, we should stop the search
|
||||
@ -95,11 +97,21 @@ func (q *QLogFile) Seek(timestamp int64) (int64, int, error) {
|
||||
|
||||
for {
|
||||
// Get the line at the specified position
|
||||
line, lineIdx, err = q.readProbeLine(probe)
|
||||
line, lineIdx, lineEndIdx, err = q.readProbeLine(probe)
|
||||
if err != nil {
|
||||
return 0, depth, err
|
||||
}
|
||||
|
||||
if lineIdx < start || lineEndIdx > end || lineIdx == lastProbeLineIdx {
|
||||
// If we're testing the same line twice then most likely
|
||||
// the scope is too narrow and we won't find anything anymore
|
||||
log.Error("querylog: didn't find timestamp:%v", timestamp)
|
||||
return 0, depth, ErrSeekNotFound
|
||||
}
|
||||
|
||||
// Save the last found idx
|
||||
lastProbeLineIdx = lineIdx
|
||||
|
||||
// Get the timestamp from the query log record
|
||||
ts := readQLogTimestamp(line)
|
||||
|
||||
@ -112,27 +124,17 @@ func (q *QLogFile) Seek(timestamp int64) (int64, int, error) {
|
||||
break
|
||||
}
|
||||
|
||||
if lastProbeLineIdx == lineIdx {
|
||||
// If we're testing the same line twice then most likely
|
||||
// the scope is too narrow and we won't find anything anymore
|
||||
return 0, depth, ErrSeekNotFound
|
||||
}
|
||||
|
||||
// Save the last found idx
|
||||
lastProbeLineIdx = lineIdx
|
||||
|
||||
// Narrow the scope and repeat the search
|
||||
if ts > timestamp {
|
||||
// If the timestamp we're looking for is OLDER than what we found
|
||||
// Then the line is somewhere on the LEFT side from the current probe position
|
||||
end = probe
|
||||
probe = start + (end-start)/2
|
||||
end = lineIdx
|
||||
} else {
|
||||
// If the timestamp we're looking for is NEWER than what we found
|
||||
// Then the line is somewhere on the RIGHT side from the current probe position
|
||||
start = probe
|
||||
probe = start + (end-start)/2
|
||||
start = lineEndIdx
|
||||
}
|
||||
probe = start + (end-start)/2
|
||||
|
||||
depth++
|
||||
if depth >= 100 {
|
||||
@ -267,7 +269,7 @@ func (q *QLogFile) initBuffer(position int64) error {
|
||||
// readProbeLine reads a line that includes the specified position
|
||||
// this method is supposed to be used when we use binary search in the Seek method
|
||||
// in the case of consecutive reads, use readNext (it uses a better buffer)
|
||||
func (q *QLogFile) readProbeLine(position int64) (string, int64, error) {
|
||||
func (q *QLogFile) readProbeLine(position int64) (string, int64, int64, error) {
|
||||
// First of all, we should read a buffer that will include the query log line
|
||||
// In order to do this, we'll define the boundaries
|
||||
seekPosition := int64(0)
|
||||
@ -280,14 +282,14 @@ func (q *QLogFile) readProbeLine(position int64) (string, int64, error) {
|
||||
// Seek to this position
|
||||
_, err := q.file.Seek(seekPosition, io.SeekStart)
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
return "", 0, 0, err
|
||||
}
|
||||
|
||||
// The buffer size is 2*maxEntrySize
|
||||
buffer := make([]byte, maxEntrySize*2)
|
||||
bufferLen, err := q.file.Read(buffer)
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
return "", 0, 0, err
|
||||
}
|
||||
|
||||
// Now start looking for the new line character starting
|
||||
@ -301,16 +303,18 @@ func (q *QLogFile) readProbeLine(position int64) (string, int64, error) {
|
||||
}
|
||||
// Looking for the end of line now
|
||||
var endLine = int64(bufferLen)
|
||||
lineEndIdx := endLine + seekPosition
|
||||
for i := relativePos; i < int64(bufferLen); i++ {
|
||||
if buffer[i] == '\n' {
|
||||
endLine = i
|
||||
lineEndIdx = endLine + seekPosition + 1
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Finally we can return the string we were looking for
|
||||
lineIdx := startLine + seekPosition
|
||||
return string(buffer[startLine:endLine]), lineIdx, nil
|
||||
return string(buffer[startLine:endLine]), lineIdx, lineEndIdx, nil
|
||||
}
|
||||
|
||||
// readQLogTimestamp reads the timestamp field from the query log line
|
||||
|
||||
@ -266,3 +266,25 @@ func prepareTestFiles(dir string, filesCount, linesCount int) []string {
|
||||
|
||||
return files
|
||||
}
|
||||
|
||||
func TestQLogSeek(t *testing.T) {
|
||||
testDir := prepareTestDir()
|
||||
defer func() { _ = os.RemoveAll(testDir) }()
|
||||
|
||||
d := `{"T":"2020-08-31T18:44:23.911246629+03:00","QH":"wfqvjymurpwegyv","QT":"A","QC":"IN","CP":"","Answer":"","Result":{},"Elapsed":66286385,"Upstream":"tls://dns-unfiltered.adguard.com:853"}
|
||||
{"T":"2020-08-31T18:44:25.376690873+03:00"}
|
||||
{"T":"2020-08-31T18:44:25.382540454+03:00"}`
|
||||
f, _ := ioutil.TempFile(testDir, "*.txt")
|
||||
_, _ = f.WriteString(d)
|
||||
defer f.Close()
|
||||
|
||||
q, err := NewQLogFile(f.Name())
|
||||
assert.Nil(t, err)
|
||||
defer q.Close()
|
||||
|
||||
target, _ := time.Parse(time.RFC3339, "2020-08-31T18:44:25.376690873+03:00")
|
||||
|
||||
_, depth, err := q.Seek(target.UnixNano())
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, 1, depth)
|
||||
}
|
||||
|
||||
@ -78,7 +78,7 @@ func (l *queryLog) flushToFile(buffer []*logEntry) error {
|
||||
return err
|
||||
}
|
||||
|
||||
log.Debug("ok \"%s\": %v bytes written", filename, n)
|
||||
log.Debug("querylog: ok \"%s\": %v bytes written", filename, n)
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -94,20 +94,46 @@ func (l *queryLog) rotate() error {
|
||||
|
||||
err := os.Rename(from, to)
|
||||
if err != nil {
|
||||
log.Error("Failed to rename querylog: %s", err)
|
||||
log.Error("querylog: failed to rename file: %s", err)
|
||||
return err
|
||||
}
|
||||
|
||||
log.Debug("Rotated from %s to %s successfully", from, to)
|
||||
log.Debug("querylog: renamed %s -> %s", from, to)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *queryLog) readFileFirstTimeValue() int64 {
|
||||
f, err := os.Open(l.logFile)
|
||||
if err != nil {
|
||||
return -1
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
buf := make([]byte, 500)
|
||||
r, err := f.Read(buf)
|
||||
if err != nil {
|
||||
return -1
|
||||
}
|
||||
buf = buf[:r]
|
||||
|
||||
val := readJSONValue(string(buf), "T")
|
||||
t, err := time.Parse(time.RFC3339Nano, val)
|
||||
if err != nil {
|
||||
return -1
|
||||
}
|
||||
|
||||
log.Debug("querylog: the oldest log entry: %s", val)
|
||||
return t.Unix()
|
||||
}
|
||||
|
||||
func (l *queryLog) periodicRotate() {
|
||||
for range time.Tick(time.Duration(l.conf.Interval) * 24 * time.Hour) {
|
||||
err := l.rotate()
|
||||
if err != nil {
|
||||
log.Error("Failed to rotate querylog: %s", err)
|
||||
// do nothing, continue rotating
|
||||
intervalSeconds := uint64(l.conf.Interval) * 24 * 60 * 60
|
||||
for {
|
||||
oldest := l.readFileFirstTimeValue()
|
||||
if uint64(oldest)+intervalSeconds <= uint64(time.Now().Unix()) {
|
||||
_ = l.rotate()
|
||||
}
|
||||
|
||||
time.Sleep(24 * time.Hour)
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user