Merge pull request #2 from ndmsystems/BUILD-104

[BUILD-104] resolver with TTL

.gitignore

@@ -10,3 +10,6 @@
 
 # Output of the go coverage tool, specifically when used with LiteIDE
 *.out
+
+# IDE
+.idea

dns_client.go

@@ -0,0 +1,210 @@
+package resolver
+
+import (
+	"context"
+	"log"
+	"math"
+	"net"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/miekg/dns"
+	logApi "github.com/woody-ltd/go/api/log"
+	"golang.org/x/sync/errgroup"
+)
+
+const (
+	defaultTtl = 60 // 60 sec
+)
+
+// iDnsClient ...
+type iDnsClient interface {
+	setNameServers(nameServers []string)
+	lookupHost(ctx context.Context, host string) ([]net.IP, []net.IP, uint32, error)
+}
+
+// dnsClient ...
+type dnsClient struct {
+	sync.RWMutex
+	nsCounter   uint64
+	nameServers []string
+	logger      logApi.Logger
+}
+
+// newDnsClient ...
+func newDnsClient(logger logApi.Logger) *dnsClient {
+	return &dnsClient{
+		logger: logger,
+	}
+}
+
+// setNameServers ...
+func (d *dnsClient) setNameServers(nameServers []string) {
+	ns := parseNameServers(nameServers)
+	d.Lock()
+	defer d.Unlock()
+	d.nameServers = ns
+}
+
+// lookupHost ...
+func (d *dnsClient) lookupHost(ctx context.Context, host string) ([]net.IP, []net.IP, uint32, error) {
+	d.RLock()
+	nsCnt := len(d.nameServers)
+	d.RUnlock()
+
+	if nsCnt == 0 {
+		ips := make(map[bool][]net.IP)
+		addrs, err := net.LookupHost(host)
+		if err != nil {
+			return nil, nil, defaultTtl, nil
+		}
+		for _, addr := range addrs {
+			if netIP := net.ParseIP(addr); netIP != nil {
+				isV6 := strings.Contains(addr, ":")
+				ips[isV6] = append(ips[isV6], netIP)
+			}
+		}
+		return ips[false], ips[true], defaultTtl, nil
+	}
+
+	var (
+		ip4, ip6 []net.IP
+		ttl      uint32
+		err      error
+	)
+
+	for i := 0; i < nsCnt; i++ {
+		d.RLock()
+		nsIdx := int(atomic.LoadUint64(&d.nsCounter)) % len(d.nameServers)
+		nServer := d.nameServers[nsIdx]
+		d.RUnlock()
+
+		ip4, ip6, ttl, err = d.dnsLookupHost(ctx, nServer, host)
+		if err == nil {
+			break
+		}
+		atomic.AddUint64(&d.nsCounter, 1)
+	}
+
+	return ip4, ip6, ttl, err
+}
+
+// dnsLookupHost ...
+func (d *dnsClient) dnsLookupHost(ctx context.Context, nServer, host string) ([]net.IP, []net.IP, uint32, error) {
+	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
+	defer cancel()
+
+	var ip4, ip6 []net.IP
+	var ttl4, ttl6 uint32 = math.MaxUint32, math.MaxUint32
+
+	g, _ := errgroup.WithContext(ctx)
+
+	// get IPv4 addresses
+	g.Go(func() error {
+		m := new(dns.Msg)
+		m.SetQuestion(dns.Fqdn(host), dns.TypeA)
+		in, err := dns.Exchange(m, nServer+":53")
+		if err != nil {
+			return err
+		}
+		for _, rr := range in.Answer {
+			if dnsRec, ok := rr.(*dns.A); ok {
+				ip4 = append(ip4, dnsRec.A)
+				if dnsRec.Header().Ttl < ttl4 {
+					ttl4 = dnsRec.Header().Ttl
+				}
+			}
+		}
+		return nil
+	})
+
+	// get IPv6 addresses
+	g.Go(func() error {
+		m := new(dns.Msg)
+		m.SetQuestion(dns.Fqdn(host), dns.TypeAAAA)
+		in, err := dns.Exchange(m, nServer+":53")
+		if err != nil {
+			return err
+		}
+		for _, rr := range in.Answer {
+			if dnsRec, ok := rr.(*dns.AAAA); ok {
+				ip6 = append(ip6, dnsRec.AAAA)
+				if dnsRec.Header().Ttl < ttl6 {
+					ttl6 = dnsRec.Header().Ttl
+				}
+			}
+		}
+		return nil
+	})
+
+	var ttl uint32 = defaultTtl
+	if ttl4 > defaultTtl && ttl4 != math.MaxUint32 {
+		ttl = ttl4
+	}
+	if ttl6 > defaultTtl && ttl6 != math.MaxUint32 && ttl6 < ttl4 {
+		ttl = ttl6
+	}
+
+	return ip4, ip6, ttl, g.Wait()
+}
+
+// LookupSRV ...
+func (d *dnsClient) lookupSRV(service, proto, name string) (string, []*net.SRV, error) {
+	d.RLock()
+	nsCnt := len(d.nameServers)
+	d.RUnlock()
+
+	if nsCnt == 0 {
+		return net.LookupSRV(service, proto, name)
+	}
+
+	var (
+		cname string
+		srvs  []*net.SRV
+		err   error
+	)
+
+	for i := 0; i < nsCnt; i++ {
+		d.RLock()
+		nsIdx := int(atomic.LoadUint64(&d.nsCounter)) % len(d.nameServers)
+		nServer := d.nameServers[nsIdx]
+		d.RUnlock()
+
+		cname, srvs, err = d.dnsLookupSRV(nServer, service, proto, name)
+		if err == nil {
+			break
+		}
+		atomic.AddUint64(&d.nsCounter, 1)
+	}
+
+	return cname, srvs, err
+}
+
+func (d *dnsClient) dnsLookupSRV(nServer, service, proto, name string) (string, []*net.SRV, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	r := &net.Resolver{
+		PreferGo: true,
+		Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
+			d := net.Dialer{Timeout: 2 * time.Second}
+			return d.DialContext(ctx, network, nServer+":53")
+		},
+	}
+	return r.LookupSRV(ctx, service, proto, name)
+}
+
+// parseNameServers ...
+func parseNameServers(nameServers []string) []string {
+	ret := make([]string, 0, len(nameServers))
+	for _, ns := range nameServers {
+		if addr := net.ParseIP(ns); addr == nil {
+			log.Printf("nameserver %s is not valid\n", ns)
+			continue
+		}
+		ret = append(ret, ns)
+	}
+	return ret
+}

go.mod

@@ -2,4 +2,8 @@ module github.com/ndmsystems/go-dns-caching-resolver
 
 go 1.13
 
-require github.com/tdx/go v0.3.8
+require (
+	github.com/miekg/dns v1.1.50
+	github.com/woody-ltd/go v0.3.9
+	golang.org/x/sync v0.1.0
+)

go.sum

@@ -1,2 +1,38 @@
-github.com/tdx/go v0.3.8 h1:UC3D8TkS4XERiR3JfLcm5LOrZdupOKBSGifQeBVQEhA=
-github.com/tdx/go v0.3.8/go.mod h1:+484gBjtsvwIg11Z1wZQINKCpZJHsZ3ndnjsipY7LvI=
+github.com/miekg/dns v1.1.50 h1:DQUfb9uc6smULcREF09Uc+/Gd46YWqJd5DbpPE9xkcA=
+github.com/miekg/dns v1.1.50/go.mod h1:e3IlAVfNqAllflbibAZEWOXOQ+Ynzk/dDozDxY7XnME=
+github.com/woody-ltd/go v0.3.9 h1:7F5jWIG0BdgamVR79jPguObA9M6lBdCStrEsXiptsxU=
+github.com/woody-ltd/go v0.3.9/go.mod h1:Rn92iTTDwWE+3ulIilmIlWnBECB5QZcCm9z6UnYM2TM=
+github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/mod v0.4.2 h1:Gz96sIWK3OalVv/I/qNygP42zyoKp3xptRVCWRFEBvo=
+golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
+golang.org/x/net v0.0.0-20210726213435-c6fcb2dbf985 h1:4CSI6oo7cOjJKajidEljs9h+uP0rRZBPPPhcCbj5mw8=
+golang.org/x/net v0.0.0-20210726213435-c6fcb2dbf985/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c h1:F1jZWGFhYfh0Ci55sIpILtKKK8p3i2/krTr0H1rg74I=
+golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.1.6-0.20210726203631-07bc1bf47fb2 h1:BonxutuHCTL0rBDnZlKjpGIQFTjyUVTexFOdWkB6Fg0=
+golang.org/x/tools v0.1.6-0.20210726203631-07bc1bf47fb2/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

host.go

@@ -0,0 +1,127 @@
+package resolver
+
+import (
+	"context"
+	"net"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	logApi "github.com/woody-ltd/go/api/log"
+)
+
+const (
+	oldHostDuration  = 30 * time.Minute
+	retryIntervalSec = 10
+)
+
+// host ...
+type host struct {
+	tag      string
+	hostName string
+	ip4      *ips
+	ip6      *ips
+	lastTime int64
+
+	// eaFlag - flag means explicitly added host
+	eaFlag bool
+
+	dnsClient *dnsClient
+	logger    logApi.Logger
+
+	ready  sync.WaitGroup
+	stopCh chan struct{}
+}
+
+// newHost ...
+func newHost(tag string, hName string, eaFlag bool, dnsClient *dnsClient, logger logApi.Logger) *host {
+	h := &host{
+		tag:       tag,
+		hostName:  hName,
+		eaFlag:    eaFlag,
+		ip4:       newIps(),
+		ip6:       newIps(),
+		lastTime:  time.Now().Unix(),
+		dnsClient: dnsClient,
+		logger:    logger,
+		stopCh:    make(chan struct{}),
+	}
+
+	h.ready.Add(1)
+	go h.reloadIPsLoop()
+
+	return h
+}
+
+// getNextIP4WithIndex ...
+func (h *host) getNextIP4WithIndex() (net.IP, int) {
+	h.ready.Wait()
+	defer h.updLastTime()
+	return h.ip4.getNextIPWithIndex()
+}
+
+// getNextIP6WithIndex ...
+func (h *host) getNextIP6WithIndex() (net.IP, int) {
+	h.ready.Wait()
+	defer h.updLastTime()
+	return h.ip6.getNextIPWithIndex()
+}
+
+// getIPs ...
+func (h *host) getIPs() ([]net.IP, []net.IP) {
+	h.ready.Wait()
+	return h.ip4.getList(), h.ip6.getList()
+}
+
+// reloadIPsLoop ...
+func (h *host) reloadIPsLoop() {
+	ttl := h.reloadIPs()
+	h.ready.Done()
+
+	ttlCh := time.After(time.Duration(ttl) * time.Second)
+	for {
+		select {
+		case <-h.stopCh:
+			h.logger.Info().Println(h.tag, "Stop resolving host", h.hostName)
+			return
+		case <-ttlCh:
+			ttl = h.reloadIPs()
+			ttlCh = time.After(time.Duration(ttl) * time.Second)
+		}
+	}
+}
+
+// reloadIPs ...
+func (h *host) reloadIPs() uint32 {
+	ip4, ip6, ttl, err := h.dnsClient.lookupHost(context.Background(), h.hostName)
+	if err != nil {
+		h.logger.Error().Println(h.tag, "Error reloading ips for host", h.hostName, err)
+		return retryIntervalSec
+	}
+
+	h.ip4.setIpList(ip4)
+	h.ip6.setIpList(ip6)
+
+	return ttl
+}
+
+// isOld ...
+func (h *host) isOld() bool {
+	lastTime := atomic.LoadInt64(&h.lastTime)
+	return lastTime < time.Now().Add(-oldHostDuration).Unix()
+}
+
+// isExplicitlyAdded ...
+func (h *host) isExplicitlyAdded() bool {
+	return h.eaFlag
+}
+
+// stop ...
+func (h *host) stop() {
+	close(h.stopCh)
+}
+
+// updLastTime ...
+func (h *host) updLastTime() {
+	atomic.StoreInt64(&h.lastTime, time.Now().Unix())
+}