Pull request 1797: AG-21072-querylog-conf-race

Merge in DNS/adguard-home from AG-21072-querylog-conf-race to master

Squashed commit of the following:

commit fcb14353ee63f582986e18affebf5ed965c3bfc7
Author: Ainar Garipov <A.Garipov@AdGuard.COM>
Date:   Mon Apr 3 15:04:03 2023 +0300

    querylog: imp code, docs

commit 9070bc1d4eee5efc5795466b2c2a40c6ab495e68
Author: Ainar Garipov <A.Garipov@AdGuard.COM>
Date:   Mon Apr 3 13:58:56 2023 +0300

    querylog: fix races

internal/querylog/decode.go

@@ -166,86 +166,6 @@ var logEntryHandlers = map[string]logEntryHandler{
 	},
 }
 
-var resultHandlers = map[string]logEntryHandler{
-	"IsFiltered": func(t json.Token, ent *logEntry) error {
-		v, ok := t.(bool)
-		if !ok {
-			return nil
-		}
-		ent.Result.IsFiltered = v
-		return nil
-	},
-	"Rule": func(t json.Token, ent *logEntry) error {
-		s, ok := t.(string)
-		if !ok {
-			return nil
-		}
-
-		l := len(ent.Result.Rules)
-		if l == 0 {
-			ent.Result.Rules = []*filtering.ResultRule{{}}
-			l++
-		}
-
-		ent.Result.Rules[l-1].Text = s
-
-		return nil
-	},
-	"FilterID": func(t json.Token, ent *logEntry) error {
-		n, ok := t.(json.Number)
-		if !ok {
-			return nil
-		}
-
-		i, err := n.Int64()
-		if err != nil {
-			return err
-		}
-
-		l := len(ent.Result.Rules)
-		if l == 0 {
-			ent.Result.Rules = []*filtering.ResultRule{{}}
-			l++
-		}
-
-		ent.Result.Rules[l-1].FilterListID = i
-
-		return nil
-	},
-	"Reason": func(t json.Token, ent *logEntry) error {
-		v, ok := t.(json.Number)
-		if !ok {
-			return nil
-		}
-		i, err := v.Int64()
-		if err != nil {
-			return err
-		}
-		ent.Result.Reason = filtering.Reason(i)
-		return nil
-	},
-	"ServiceName": func(t json.Token, ent *logEntry) error {
-		s, ok := t.(string)
-		if !ok {
-			return nil
-		}
-
-		ent.Result.ServiceName = s
-
-		return nil
-	},
-	"CanonName": func(t json.Token, ent *logEntry) error {
-		s, ok := t.(string)
-		if !ok {
-			return nil
-		}
-
-		ent.Result.CanonName = s
-
-		return nil
-	},
-}
-
 func decodeResultRuleKey(key string, i int, dec *json.Decoder, ent *logEntry) {
 	var vToken json.Token
 	switch key {
@@ -582,25 +502,11 @@ func decodeResult(dec *json.Decoder, ent *logEntry) {
 			return
 		}
 
-		switch key {
+		decHandler, ok := resultDecHandlers[key]
-		case "ReverseHosts":
+		if ok {
-			decodeResultReverseHosts(dec, ent)
+			decHandler(dec, ent)
 
 			continue
-		case "IPList":
-			decodeResultIPList(dec, ent)
-
-			continue
-		case "Rules":
-			decodeResultRules(dec, ent)
-
-			continue
-		case "DNSRewriteResult":
-			decodeResultDNSRewriteResult(dec, ent)
-
-			continue
-		default:
-			// Go on.
 		}
 
 		handler, ok := resultHandlers[key]
@@ -621,6 +527,93 @@ func decodeResult(dec *json.Decoder, ent *logEntry) {
 	}
 }
 
+var resultHandlers = map[string]logEntryHandler{
+	"IsFiltered": func(t json.Token, ent *logEntry) error {
+		v, ok := t.(bool)
+		if !ok {
+			return nil
+		}
+		ent.Result.IsFiltered = v
+		return nil
+	},
+	"Rule": func(t json.Token, ent *logEntry) error {
+		s, ok := t.(string)
+		if !ok {
+			return nil
+		}
+
+		l := len(ent.Result.Rules)
+		if l == 0 {
+			ent.Result.Rules = []*filtering.ResultRule{{}}
+			l++
+		}
+
+		ent.Result.Rules[l-1].Text = s
+
+		return nil
+	},
+	"FilterID": func(t json.Token, ent *logEntry) error {
+		n, ok := t.(json.Number)
+		if !ok {
+			return nil
+		}
+
+		i, err := n.Int64()
+		if err != nil {
+			return err
+		}
+
+		l := len(ent.Result.Rules)
+		if l == 0 {
+			ent.Result.Rules = []*filtering.ResultRule{{}}
+			l++
+		}
+
+		ent.Result.Rules[l-1].FilterListID = i
+
+		return nil
+	},
+	"Reason": func(t json.Token, ent *logEntry) error {
+		v, ok := t.(json.Number)
+		if !ok {
+			return nil
+		}
+		i, err := v.Int64()
+		if err != nil {
+			return err
+		}
+		ent.Result.Reason = filtering.Reason(i)
+		return nil
+	},
+	"ServiceName": func(t json.Token, ent *logEntry) error {
+		s, ok := t.(string)
+		if !ok {
+			return nil
+		}
+
+		ent.Result.ServiceName = s
+
+		return nil
+	},
+	"CanonName": func(t json.Token, ent *logEntry) error {
+		s, ok := t.(string)
+		if !ok {
+			return nil
+		}
+
+		ent.Result.CanonName = s
+
+		return nil
+	},
+}
+
+var resultDecHandlers = map[string]func(dec *json.Decoder, ent *logEntry){
+	"ReverseHosts":     decodeResultReverseHosts,
+	"IPList":           decodeResultIPList,
+	"Rules":            decodeResultRules,
+	"DNSRewriteResult": decodeResultDNSRewriteResult,
+}
+
 func decodeLogEntry(ent *logEntry, str string) {
 	dec := json.NewDecoder(strings.NewReader(str))
 	dec.UseNumber()

internal/querylog/entry.go

@@ -0,0 +1,70 @@
+package querylog
+
+import (
+	"net"
+	"time"
+
+	"github.com/AdguardTeam/AdGuardHome/internal/filtering"
+	"github.com/AdguardTeam/golibs/errors"
+	"github.com/AdguardTeam/golibs/log"
+	"github.com/miekg/dns"
+)
+
+// logEntry represents a single entry in the file.
+type logEntry struct {
+	// client is the found client information, if any.
+	client *Client
+
+	Time time.Time `json:"T"`
+
+	QHost  string `json:"QH"`
+	QType  string `json:"QT"`
+	QClass string `json:"QC"`
+
+	ReqECS string `json:"ECS,omitempty"`
+
+	ClientID    string      `json:"CID,omitempty"`
+	ClientProto ClientProto `json:"CP"`
+
+	Upstream string `json:",omitempty"`
+
+	Answer     []byte `json:",omitempty"`
+	OrigAnswer []byte `json:",omitempty"`
+
+	IP net.IP `json:"IP"`
+
+	Result filtering.Result
+
+	Elapsed time.Duration
+
+	Cached            bool `json:",omitempty"`
+	AuthenticatedData bool `json:"AD,omitempty"`
+}
+
+// shallowClone returns a shallow clone of e.
+func (e *logEntry) shallowClone() (clone *logEntry) {
+	cloneVal := *e
+
+	return &cloneVal
+}
+
+// addResponse adds data from resp to e.Answer if resp is not nil.  If isOrig is
+// true, addResponse sets the e.OrigAnswer field instead of e.Answer.  Any
+// errors are logged.
+func (e *logEntry) addResponse(resp *dns.Msg, isOrig bool) {
+	if resp == nil {
+		return
+	}
+
+	var err error
+	if isOrig {
+		e.Answer, err = resp.Pack()
+		err = errors.Annotate(err, "packing answer: %w")
+	} else {
+		e.OrigAnswer, err = resp.Pack()
+		err = errors.Annotate(err, "packing orig answer: %w")
+	}
+	if err != nil {
+		log.Error("querylog: %s", err)
+	}
+}

internal/querylog/qlog.go

@@ -3,7 +3,6 @@ package querylog
 
 import (
 	"fmt"
-	"net"
 	"os"
 	"strings"
 	"sync"
@@ -13,6 +12,7 @@ import (
 	"github.com/AdguardTeam/AdGuardHome/internal/filtering"
 	"github.com/AdguardTeam/golibs/errors"
 	"github.com/AdguardTeam/golibs/log"
+	"github.com/AdguardTeam/golibs/stringutil"
 	"github.com/AdguardTeam/golibs/timeutil"
 	"github.com/miekg/dns"
 )
@@ -74,52 +74,24 @@ func NewClientProto(s string) (cp ClientProto, err error) {
 	}
 }
 
-// logEntry - represents a single log entry
-type logEntry struct {
-	// client is the found client information, if any.
-	client *Client
-
-	Time time.Time `json:"T"`
-
-	QHost  string `json:"QH"`
-	QType  string `json:"QT"`
-	QClass string `json:"QC"`
-
-	ReqECS string `json:"ECS,omitempty"`
-
-	ClientID    string      `json:"CID,omitempty"`
-	ClientProto ClientProto `json:"CP"`
-
-	Answer     []byte `json:",omitempty"` // sometimes empty answers happen like binerdunt.top or rev2.globalrootservers.net
-	OrigAnswer []byte `json:",omitempty"`
-
-	Result   filtering.Result
-	Upstream string `json:",omitempty"`
-
-	IP net.IP `json:"IP"`
-
-	Elapsed time.Duration
-
-	Cached            bool `json:",omitempty"`
-	AuthenticatedData bool `json:"AD,omitempty"`
-}
-
-// shallowClone returns a shallow clone of e.
-func (e *logEntry) shallowClone() (clone *logEntry) {
-	cloneVal := *e
-
-	return &cloneVal
-}
-
 func (l *queryLog) Start() {
 	if l.conf.HTTPRegister != nil {
 		l.initWeb()
 	}
+
 	go l.periodicRotate()
 }
 
 func (l *queryLog) Close() {
-	_ = l.flushLogBuffer(true)
+	l.confMu.RLock()
+	defer l.confMu.RUnlock()
+
+	if l.conf.FileEnabled {
+		err := l.flushLogBuffer()
+		if err != nil {
+			log.Error("querylog: closing: %s", err)
+		}
+	}
 }
 
 func checkInterval(ivl time.Duration) (ok bool) {
@@ -150,7 +122,13 @@ func validateIvl(ivl time.Duration) (err error) {
 }
 
 func (l *queryLog) WriteDiskConfig(c *Config) {
+	l.confMu.RLock()
+	defer l.confMu.RUnlock()
+
 	*c = *l.conf
+
+	// TODO(a.garipov): Add stringutil.Set.Clone.
+	c.Ignored = stringutil.NewSet(l.conf.Ignored.Values()...)
 }
 
 // Clear memory buffer and remove log files
@@ -181,7 +159,17 @@ func (l *queryLog) clear() {
 }
 
 func (l *queryLog) Add(params *AddParams) {
-	if !l.conf.Enabled {
+	var isEnabled, fileIsEnabled bool
+	var memSize uint32
+	func() {
+		l.confMu.RLock()
+		defer l.confMu.RUnlock()
+
+		isEnabled, fileIsEnabled = l.conf.Enabled, l.conf.FileEnabled
+		memSize = l.conf.MemSize
+	}()
+
+	if !isEnabled {
 		return
 	}
 
@@ -198,7 +186,7 @@ func (l *queryLog) Add(params *AddParams) {
 
 	now := time.Now()
 	q := params.Question.Question[0]
-	entry := logEntry{
+	entry := &logEntry{
 		Time: now,
 
 		QHost:  strings.ToLower(q.Name[:len(q.Name)-1]),
@@ -223,39 +211,18 @@ func (l *queryLog) Add(params *AddParams) {
 		entry.ReqECS = params.ReqECS.String()
 	}
 
-	if params.Answer != nil {
+	entry.addResponse(params.Answer, false)
-		var a []byte
+	entry.addResponse(params.OrigAnswer, true)
-		a, err = params.Answer.Pack()
-		if err != nil {
-			log.Error("querylog: Answer.Pack(): %s", err)
-
-			return
-		}
-
-		entry.Answer = a
-	}
-
-	if params.OrigAnswer != nil {
-		var a []byte
-		a, err = params.OrigAnswer.Pack()
-		if err != nil {
-			log.Error("querylog: OrigAnswer.Pack(): %s", err)
-
-			return
-		}
-
-		entry.OrigAnswer = a
-	}
 
 	needFlush := false
 	func() {
 		l.bufferLock.Lock()
 		defer l.bufferLock.Unlock()
 
-		l.buffer = append(l.buffer, &entry)
+		l.buffer = append(l.buffer, entry)
 
-		if !l.conf.FileEnabled {
+		if !fileIsEnabled {
-			if len(l.buffer) > int(l.conf.MemSize) {
+			if len(l.buffer) > int(memSize) {
 				// Writing to file is disabled, so just remove the oldest entry
 				// from the slices.
 				//
@@ -265,17 +232,19 @@ func (l *queryLog) Add(params *AddParams) {
 				l.buffer = l.buffer[1:]
 			}
 		} else if !l.flushPending {
-			needFlush = len(l.buffer) >= int(l.conf.MemSize)
+			needFlush = len(l.buffer) >= int(memSize)
 			if needFlush {
 				l.flushPending = true
 			}
 		}
 	}()
 
-	// if buffer needs to be flushed to disk, do it now
 	if needFlush {
 		go func() {
-			_ = l.flushLogBuffer(false)
+			flushErr := l.flushLogBuffer()
+			if flushErr != nil {
+				log.Error("querylog: flushing after adding: %s", err)
+			}
 		}()
 	}
 }
@@ -288,7 +257,8 @@ func (l *queryLog) ShouldLog(host string, _, _ uint16) bool {
 	return !l.isIgnored(host)
 }
 
-// isIgnored returns true if the host is in the Ignored list.
+// isIgnored returns true if the host is in the ignored domains list.  It
+// assumes that l.confMu is locked for reading.
 func (l *queryLog) isIgnored(host string) bool {
 	return l.conf.Ignored.Has(host)
 }

internal/querylog/qlog_test.go

@@ -34,13 +34,13 @@ func TestQueryLog(t *testing.T) {
 	// Add disk entries.
 	addEntry(l, "example.org", net.IPv4(1, 1, 1, 1), net.IPv4(2, 2, 2, 1))
 	// Write to disk (first file).
-	require.NoError(t, l.flushLogBuffer(true))
+	require.NoError(t, l.flushLogBuffer())
 	// Start writing to the second file.
 	require.NoError(t, l.rotate())
 	// Add disk entries.
 	addEntry(l, "example.org", net.IPv4(1, 1, 1, 2), net.IPv4(2, 2, 2, 2))
 	// Write to disk.
-	require.NoError(t, l.flushLogBuffer(true))
+	require.NoError(t, l.flushLogBuffer())
 	// Add memory entries.
 	addEntry(l, "test.example.org", net.IPv4(1, 1, 1, 3), net.IPv4(2, 2, 2, 3))
 	addEntry(l, "example.com", net.IPv4(1, 1, 1, 4), net.IPv4(2, 2, 2, 4))
@@ -144,7 +144,7 @@ func TestQueryLogOffsetLimit(t *testing.T) {
 		addEntry(l, secondPageDomain, net.IPv4(1, 1, 1, 1), net.IPv4(2, 2, 2, 1))
 	}
 	// Write them to the first file.
-	require.NoError(t, l.flushLogBuffer(true))
+	require.NoError(t, l.flushLogBuffer())
 	// Add more to the in-memory part of log.
 	for i := 0; i < entNum; i++ {
 		addEntry(l, firstPageDomain, net.IPv4(1, 1, 1, 1), net.IPv4(2, 2, 2, 1))
@@ -216,7 +216,7 @@ func TestQueryLogMaxFileScanEntries(t *testing.T) {
 		addEntry(l, "example.org", net.IPv4(1, 1, 1, 1), net.IPv4(2, 2, 2, 1))
 	}
 	// Write them to disk.
-	require.NoError(t, l.flushLogBuffer(true))
+	require.NoError(t, l.flushLogBuffer())
 
 	params := newSearchParams()
 

internal/querylog/querylogfile.go

@@ -13,40 +13,23 @@ import (
 
 // flushLogBuffer flushes the current buffer to file and resets the current
 // buffer.
-func (l *queryLog) flushLogBuffer(fullFlush bool) (err error) {
+func (l *queryLog) flushLogBuffer() (err error) {
-	if !l.conf.FileEnabled {
-		return nil
-	}
-
 	l.fileFlushLock.Lock()
 	defer l.fileFlushLock.Unlock()
 
-	// Flush the remainder to file.
 	var flushBuffer []*logEntry
-	needFlush := fullFlush
 	func() {
 		l.bufferLock.Lock()
 		defer l.bufferLock.Unlock()
 
-		needFlush = needFlush || len(l.buffer) >= int(l.conf.MemSize)
+		flushBuffer = l.buffer
-		if needFlush {
+		l.buffer = nil
-			flushBuffer = l.buffer
+		l.flushPending = false
-			l.buffer = nil
-			l.flushPending = false
-		}
 	}()
-	if !needFlush {
-		return nil
-	}
 
 	err = l.flushToFile(flushBuffer)
-	if err != nil {
-		log.Error("querylog: writing to file: %s", err)
 
-		return err
+	return errors.Annotate(err, "writing to file: %w")
-	}
-
-	return nil
 }
 
 // flushToFile saves the specified log entries to the query log file
@@ -167,8 +150,13 @@ func (l *queryLog) periodicRotate() {
 // checkAndRotate rotates log files if those are older than the specified
 // rotation interval.
 func (l *queryLog) checkAndRotate() {
-	l.confMu.RLock()
+	var rotationIvl time.Duration
-	defer l.confMu.RUnlock()
+	func() {
+		l.confMu.RLock()
+		defer l.confMu.RUnlock()
+
+		rotationIvl = l.conf.RotationIvl
+	}()
 
 	oldest, err := l.readFileFirstTimeValue()
 	if err != nil && !errors.Is(err, os.ErrNotExist) {
@@ -177,11 +165,11 @@ func (l *queryLog) checkAndRotate() {
 		return
 	}
 
-	if rot, now := oldest.Add(l.conf.RotationIvl), time.Now(); rot.After(now) {
+	if rotTime, now := oldest.Add(rotationIvl), time.Now(); rotTime.After(now) {
 		log.Debug(
 			"querylog: %s <= %s, not rotating",
 			now.Format(time.RFC3339),
-			rot.Format(time.RFC3339),
+			rotTime.Format(time.RFC3339),
 		)
 
 		return

scripts/make/go-lint.sh

@@ -161,10 +161,10 @@ run_linter "$GO" vet ./...
 run_linter govulncheck ./...
 
 # Apply more lax standards to the code we haven't properly refactored yet.
-run_linter gocyclo --over 14 ./internal/querylog/
 run_linter gocyclo --over 13\
 	./internal/dhcpd\
 	./internal/home/\
+	./internal/querylog/\
 	;
 
 # Apply the normal standards to new or somewhat refactored code.