Pull request 2291: AGDNS-2374-slog-client
Squashed commit of the following:
commit e8e6dba18b8f44392bd88999e481723a00aa3042
Merge: 929283702 41cce6259
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Tue Oct 22 13:46:26 2024 +0300
Merge branch 'master' into AGDNS-2374-slog-client
commit 929283702a6a82163906e624f965b934e3b8074e
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Tue Oct 15 14:30:00 2024 +0300
client: imp tests
commit f29d8edb89e3f05e0bd9c3b0eccc0587882a3ed3
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Mon Oct 14 15:03:08 2024 +0300
client: imp docs
commit 0b4311ac26c704bbfa0edfc51767c9fc74f959a2
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Fri Oct 11 19:12:50 2024 +0300
all: imp code
commit 1ad99ee3cb915bb19eff8b893deae1dd4d64b190
Author: Stanislav Chzhen <s.chzhen@adguard.com>
Date: Thu Oct 10 20:59:46 2024 +0300
all: slog client
This commit is contained in:
@@ -11,7 +11,6 @@ import (
|
||||
"github.com/AdguardTeam/AdGuardHome/internal/rdns"
|
||||
"github.com/AdguardTeam/AdGuardHome/internal/whois"
|
||||
"github.com/AdguardTeam/golibs/errors"
|
||||
"github.com/AdguardTeam/golibs/log"
|
||||
"github.com/AdguardTeam/golibs/logutil/slogutil"
|
||||
"github.com/AdguardTeam/golibs/netutil"
|
||||
)
|
||||
@@ -22,7 +21,7 @@ const ErrClosed errors.Error = "use of closed address processor"
|
||||
|
||||
// AddressProcessor is the interface for types that can process clients.
|
||||
type AddressProcessor interface {
|
||||
Process(ip netip.Addr)
|
||||
Process(ctx context.Context, ip netip.Addr)
|
||||
Close() (err error)
|
||||
}
|
||||
|
||||
@@ -33,7 +32,7 @@ type EmptyAddrProc struct{}
|
||||
var _ AddressProcessor = EmptyAddrProc{}
|
||||
|
||||
// Process implements the [AddressProcessor] interface for EmptyAddrProc.
|
||||
func (EmptyAddrProc) Process(_ netip.Addr) {}
|
||||
func (EmptyAddrProc) Process(_ context.Context, _ netip.Addr) {}
|
||||
|
||||
// Close implements the [AddressProcessor] interface for EmptyAddrProc.
|
||||
func (EmptyAddrProc) Close() (_ error) { return nil }
|
||||
@@ -90,12 +89,15 @@ type DefaultAddrProcConfig struct {
|
||||
type AddressUpdater interface {
|
||||
// UpdateAddress updates information about an IP address, setting host (if
|
||||
// not empty) and WHOIS information (if not nil).
|
||||
UpdateAddress(ip netip.Addr, host string, info *whois.Info)
|
||||
UpdateAddress(ctx context.Context, ip netip.Addr, host string, info *whois.Info)
|
||||
}
|
||||
|
||||
// DefaultAddrProc processes incoming client addresses with rDNS and WHOIS, if
|
||||
// configured, and updates that information in a client storage.
|
||||
type DefaultAddrProc struct {
|
||||
// logger is used to log the operation of address processor.
|
||||
logger *slog.Logger
|
||||
|
||||
// clientIPsMu serializes closure of clientIPs and access to isClosed.
|
||||
clientIPsMu *sync.Mutex
|
||||
|
||||
@@ -142,6 +144,7 @@ const (
|
||||
// not be nil.
|
||||
func NewDefaultAddrProc(c *DefaultAddrProcConfig) (p *DefaultAddrProc) {
|
||||
p = &DefaultAddrProc{
|
||||
logger: c.BaseLogger.With(slogutil.KeyPrefix, "addrproc"),
|
||||
clientIPsMu: &sync.Mutex{},
|
||||
clientIPs: make(chan netip.Addr, defaultQueueSize),
|
||||
rdns: &rdns.Empty{},
|
||||
@@ -164,10 +167,13 @@ func NewDefaultAddrProc(c *DefaultAddrProcConfig) (p *DefaultAddrProc) {
|
||||
p.whois = newWHOIS(c.BaseLogger.With(slogutil.KeyPrefix, "whois"), c.DialContext)
|
||||
}
|
||||
|
||||
go p.process(c.CatchPanics)
|
||||
// TODO(s.chzhen): Pass context.
|
||||
ctx := context.TODO()
|
||||
|
||||
go p.process(ctx, c.CatchPanics)
|
||||
|
||||
for _, ip := range c.InitialAddresses {
|
||||
p.Process(ip)
|
||||
p.Process(ctx, ip)
|
||||
}
|
||||
|
||||
return p
|
||||
@@ -210,7 +216,7 @@ func newWHOIS(logger *slog.Logger, dialFunc aghnet.DialContextFunc) (w whois.Int
|
||||
var _ AddressProcessor = (*DefaultAddrProc)(nil)
|
||||
|
||||
// Process implements the [AddressProcessor] interface for *DefaultAddrProc.
|
||||
func (p *DefaultAddrProc) Process(ip netip.Addr) {
|
||||
func (p *DefaultAddrProc) Process(ctx context.Context, ip netip.Addr) {
|
||||
p.clientIPsMu.Lock()
|
||||
defer p.clientIPsMu.Unlock()
|
||||
|
||||
@@ -222,38 +228,42 @@ func (p *DefaultAddrProc) Process(ip netip.Addr) {
|
||||
case p.clientIPs <- ip:
|
||||
// Go on.
|
||||
default:
|
||||
log.Debug("clients: ip channel is full; len: %d", len(p.clientIPs))
|
||||
p.logger.DebugContext(ctx, "ip channel is full", "len", len(p.clientIPs))
|
||||
}
|
||||
}
|
||||
|
||||
// process processes the incoming client IP-address information. It is intended
|
||||
// to be used as a goroutine. Once clientIPs is closed, process exits.
|
||||
func (p *DefaultAddrProc) process(catchPanics bool) {
|
||||
func (p *DefaultAddrProc) process(ctx context.Context, catchPanics bool) {
|
||||
if catchPanics {
|
||||
defer log.OnPanic("addrProcessor.process")
|
||||
defer slogutil.RecoverAndLog(ctx, p.logger)
|
||||
}
|
||||
|
||||
log.Info("clients: processing addresses")
|
||||
|
||||
ctx := context.TODO()
|
||||
p.logger.InfoContext(ctx, "processing addresses")
|
||||
|
||||
for ip := range p.clientIPs {
|
||||
host := p.processRDNS(ctx, ip)
|
||||
info := p.processWHOIS(ctx, ip)
|
||||
|
||||
p.addrUpdater.UpdateAddress(ip, host, info)
|
||||
p.addrUpdater.UpdateAddress(ctx, ip, host, info)
|
||||
}
|
||||
|
||||
log.Info("clients: finished processing addresses")
|
||||
p.logger.InfoContext(ctx, "finished processing addresses")
|
||||
}
|
||||
|
||||
// processRDNS resolves the clients' IP addresses using reverse DNS. host is
|
||||
// empty if there were errors or if the information hasn't changed.
|
||||
func (p *DefaultAddrProc) processRDNS(ctx context.Context, ip netip.Addr) (host string) {
|
||||
start := time.Now()
|
||||
log.Debug("clients: processing %s with rdns", ip)
|
||||
p.logger.DebugContext(ctx, "processing rdns", "ip", ip)
|
||||
defer func() {
|
||||
log.Debug("clients: finished processing %s with rdns in %s", ip, time.Since(start))
|
||||
p.logger.DebugContext(
|
||||
ctx,
|
||||
"finished processing rdns",
|
||||
"ip", ip,
|
||||
"host", host,
|
||||
"elapsed", time.Since(start),
|
||||
)
|
||||
}()
|
||||
|
||||
ok := p.shouldResolve(ip)
|
||||
@@ -280,9 +290,15 @@ func (p *DefaultAddrProc) shouldResolve(ip netip.Addr) (ok bool) {
|
||||
// hasn't changed.
|
||||
func (p *DefaultAddrProc) processWHOIS(ctx context.Context, ip netip.Addr) (info *whois.Info) {
|
||||
start := time.Now()
|
||||
log.Debug("clients: processing %s with whois", ip)
|
||||
p.logger.DebugContext(ctx, "processing whois", "ip", ip)
|
||||
defer func() {
|
||||
log.Debug("clients: finished processing %s with whois in %s", ip, time.Since(start))
|
||||
p.logger.DebugContext(
|
||||
ctx,
|
||||
"finished processing whois",
|
||||
"ip", ip,
|
||||
"whois", info,
|
||||
"elapsed", time.Since(start),
|
||||
)
|
||||
}()
|
||||
|
||||
// TODO(s.chzhen): Move the timeout logic from WHOIS configuration to the
|
||||
|
||||
Reference in New Issue
Block a user