address comments

Signed-off-by: LiZhenCheng9527 <[email protected]>

Author: LiZhenCheng9527

pkg/controller/ads/dns.go

@@ -20,115 +20,119 @@ import (
 	"net"
 	"net/netip"
 	"slices"
+	"time"
 
 	clusterv3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	endpointv3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
 	"google.golang.org/protobuf/types/known/wrapperspb"
-	"k8s.io/client-go/util/workqueue"
 
 	core_v2 "kmesh.net/kmesh/api/v2/core"
 	"kmesh.net/kmesh/pkg/dns"
 )
 
 // adsDnsResolver is DNS resolver of Kernel Native
-type AdsDnsResolver struct {
-	Clusters        chan []*clusterv3.Cluster
-	adsCache        *AdsCache
-	dnsResolver     *dns.DNSResolver
-	dnsRefreshQueue workqueue.TypedDelayingInterface[any]
+type dnsController struct {
+	Clusters    chan []*clusterv3.Cluster
+	cache       *AdsCache
+	dnsResolver *dns.DNSResolver
 }
 
-func NewAdsDnsResolver(adsCache *AdsCache) (*AdsDnsResolver, error) {
+// pending resolve domain info of Kennel-Native Mode,
+// domain name is used for dns resolution
+// cluster is used for create the apicluster
+type pendingResolveDomain struct {
+	DomainName  string
+	Clusters    []*clusterv3.Cluster
+	RefreshRate time.Duration
+}
+
+func NewDnsResolver(adsCache *AdsCache) (*dnsController, error) {
 	resolver, err := dns.NewDNSResolver()
 	if err != nil {
 		return nil, err
 	}
-	return &AdsDnsResolver{
-		Clusters:        make(chan []*clusterv3.Cluster),
-		dnsRefreshQueue: workqueue.NewTypedDelayingQueueWithConfig(workqueue.TypedDelayingQueueConfig[any]{Name: "dnsRefreshQueue"}),
-		adsCache:        adsCache,
-		dnsResolver:     resolver,
+	return &dnsController{
+		Clusters: make(chan []*clusterv3.Cluster),
+		// dnsRefreshQueue: workqueue.NewTypedDelayingQueueWithConfig(workqueue.TypedDelayingQueueConfig[any]{Name: "dnsRefreshQueue"}),
+		cache:       adsCache,
+		dnsResolver: resolver,
 	}, nil
 }
 
-func (adsResolver *AdsDnsResolver) StartAdsDnsResolver(stopCh <-chan struct{}) {
-	go adsResolver.startAdsResolver()
-	go adsResolver.refreshAdsWorker()
+func (r *dnsController) StartKernelNativeDnsController(stopCh <-chan struct{}) {
+	go r.startDnsController()
+	// start dns resolver
+	go r.dnsResolver.StartDnsResolver(stopCh)
 	go func() {
 		<-stopCh
-		adsResolver.dnsRefreshQueue.ShutDown()
-		close(adsResolver.Clusters)
+		close(r.Clusters)
 	}()
 }
 
-func (adsResolver *AdsDnsResolver) startAdsResolver() {
+func (r *dnsController) startDnsController() {
 	rateLimiter := make(chan struct{}, dns.MaxConcurrency)
-	for clusters := range adsResolver.Clusters {
+	for clusters := range r.Clusters {
 		rateLimiter <- struct{}{}
 		go func(clusters []*clusterv3.Cluster) {
 			defer func() {
 				<-rateLimiter
 			}()
-			adsResolver.resolveDomains(clusters)
+			r.resolveDomains(clusters)
 		}(clusters)
 	}
 }
 
-func (adsResolver *AdsDnsResolver) refreshAdsDns() bool {
-	element, quit := adsResolver.dnsRefreshQueue.Get()
-	if quit {
-		return false
-	}
-	defer adsResolver.dnsRefreshQueue.Done(element)
-	e := element.(*dns.PendingResolveDomain)
-
-	adsResolver.dnsResolver.RLock()
-	_, exist := adsResolver.dnsResolver.Cache[e.DomainName]
-	adsResolver.dnsResolver.RUnlock()
-	// if the domain is no longer watched, no need to refresh it
-	if !exist {
-		return true
-	}
-	addresses, ttl, err := adsResolver.dnsResolver.Resolve(e.DomainName)
-	if err != nil {
-		log.Errorf("failed to dns resolve: %v", err)
-		return false
-	}
-	if ttl > e.RefreshRate {
-		ttl = e.RefreshRate
-	}
-	if ttl == 0 {
-		ttl = dns.DeRefreshInterval
-	}
-	adsResolver.dnsRefreshQueue.AddAfter(e, ttl)
+func (r *dnsController) resolveDomains(cds []*clusterv3.Cluster) {
+	domains := getPendingResolveDomain(cds)
+	hostNames := make(map[string]struct{})
 
-	adsResolver.adsDnsResolve(e, addresses)
-	adsResolver.adsCache.ClusterCache.Flush()
-	return true
-}
+	for k := range domains {
+		hostNames[k] = struct{}{}
+	}
 
-func (adsResolver *AdsDnsResolver) resolveDomains(cds []*clusterv3.Cluster) {
-	domains := getPendingResolveDomain(cds)
+	// delete any scheduled re-resolve for domains we no longer care about
+	r.dnsResolver.RemoveUnwatchDomain(hostNames)
+	// Directly update the clusters that can find the dns resolution result in the cache
+	alreadyResolveDomains := r.dnsResolver.GetAddressesFromCache(hostNames)
+	for k, v := range alreadyResolveDomains {
+		pendingDomain := domains[k]
+		r.adsDnsResolve(pendingDomain, v.Addresses)
+		r.cache.ClusterCache.Flush()
+		delete(domains, k)
+	}
 
-	// Stow domain updates, need to remove unwatched domains first
-	adsResolver.dnsResolver.RemoveUnwatchedDomain(domains)
 	for k, v := range domains {
-		adsResolver.dnsResolver.ResolveDomains(k)
-		adsResolver.dnsRefreshQueue.AddAfter(v, 0)
+		r.dnsResolver.ResolveDomains(k)
+		domainInfo := &dns.DomainInfo{
+			Domain:      v.DomainName,
+			RefreshRate: v.RefreshRate,
+		}
+		r.dnsResolver.RefreshQueue.AddAfter(domainInfo, 0)
 	}
+	go r.refreshAdsWorker(domains)
 }
 
-func (adsResolver *AdsDnsResolver) refreshAdsWorker() {
-	for adsResolver.refreshAdsDns() {
+func (r *dnsController) refreshAdsWorker(domains map[string]*pendingResolveDomain) {
+	for !(len(domains) == 0) {
+		domain := <-r.dnsResolver.AdsDnsChan
+		v, ok := domains[domain]
+		// will this happen?
+		if !ok {
+			continue
+		}
+		addresses, _ := r.dnsResolver.GetOneDomainFromCache(domain)
+		r.adsDnsResolve(v, addresses)
+		r.cache.ClusterCache.Flush()
+		delete(domains, domain)
 	}
 }
 
-func (adsResolver *AdsDnsResolver) adsDnsResolve(pendingDomain *dns.PendingResolveDomain, addrs []string) {
+func (r *dnsController) adsDnsResolve(pendingDomain *pendingResolveDomain, addrs []string) {
 	for _, cluster := range pendingDomain.Clusters {
 		ready := overwriteDnsCluster(cluster, pendingDomain.DomainName, addrs)
 		if ready {
-			if !adsResolver.adsCache.UpdateApiClusterIfExists(core_v2.ApiStatus_UPDATE, cluster) {
+			if !r.cache.UpdateApiClusterIfExists(core_v2.ApiStatus_UPDATE, cluster) {
 				log.Debugf("cluster: %s is deleted", cluster.Name)
 				return
 			}
@@ -201,8 +205,8 @@ func overwriteDnsCluster(cluster *clusterv3.Cluster, domain string, addrs []stri
 	return ready
 }
 
-func getPendingResolveDomain(cds []*clusterv3.Cluster) map[string]*dns.PendingResolveDomain {
-	domains := make(map[string]*dns.PendingResolveDomain)
+func getPendingResolveDomain(cds []*clusterv3.Cluster) map[string]*pendingResolveDomain {
+	domains := make(map[string]*pendingResolveDomain)
 
 	for _, cluster := range cds {
 		if cluster.LoadAssignment == nil {
@@ -224,7 +228,7 @@ func getPendingResolveDomain(cds []*clusterv3.Cluster) map[string]*dns.PendingRe
 				if v, ok := domains[address]; ok {
 					v.Clusters = append(v.Clusters, cluster)
 				} else {
-					domainWithRefreshRate := &dns.PendingResolveDomain{
+					domainWithRefreshRate := &pendingResolveDomain{
 						DomainName:  address,
 						Clusters:    []*clusterv3.Cluster{cluster},
 						RefreshRate: cluster.GetDnsRefreshRate().AsDuration(),

pkg/controller/ads/dns_test.go

@@ -32,7 +32,6 @@ import (
 	"istio.io/istio/pkg/test/util/retry"
 
 	core_v2 "kmesh.net/kmesh/api/v2/core"
-	"kmesh.net/kmesh/pkg/dns"
 )
 
 func TestOverwriteDNSCluster(t *testing.T) {
@@ -197,19 +196,14 @@ func TestHandleCdsResponseWithDns(t *testing.T) {
 			clusters: []*clusterv3.Cluster{cluster1, cluster2},
 			expected: []string{"foo.bar", "foo.baz"},
 		},
-		{
-			name:     "remove all DNS type clusters",
-			clusters: []*clusterv3.Cluster{},
-			expected: []string{},
-		},
 	}
 
 	p := NewController(nil).Processor
 	stopCh := make(chan struct{})
 	defer close(stopCh)
-	dnsResolver, err := NewAdsDnsResolver(p.Cache)
+	dnsResolver, err := NewDnsResolver(p.Cache)
 	assert.NoError(t, err)
-	dnsResolver.StartAdsDnsResolver(stopCh)
+	dnsResolver.StartKernelNativeDnsController(stopCh)
 	p.DnsResolverChan = dnsResolver.Clusters
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
@@ -222,132 +216,6 @@ func TestHandleCdsResponseWithDns(t *testing.T) {
 	}
 }
 
-func TestDNS(t *testing.T) {
-	fakeDNSServer := dns.NewFakeDNSServer()
-
-	testDNSResolver, err := NewAdsDnsResolver(NewAdsCache(nil))
-	if err != nil {
-		t.Fatal(err)
-	}
-	stopCh := make(chan struct{})
-	defer close(stopCh)
-	// testDNSResolver.StartAdsDnsResolver(stopCh)
-	dnsServer := fakeDNSServer.Server.PacketConn.LocalAddr().String()
-	testDNSResolver.dnsResolver.ResolvConfServers = []string{dnsServer}
-
-	testCases := []struct {
-		name             string
-		domain           string
-		refreshRate      time.Duration
-		ttl              time.Duration
-		expected         []string
-		expectedAfterTTL []string
-		registerDomain   func(domain string)
-	}{
-		{
-			name:        "success",
-			domain:      "www.google.com.",
-			refreshRate: 10 * time.Second,
-			expected:    []string{"10.0.0.1", "fd00::1"},
-			registerDomain: func(domain string) {
-				fakeDNSServer.SetHosts(domain, 1)
-			},
-		},
-		{
-			name:             "check dns refresh after ttl, ttl < refreshRate",
-			domain:           "www.bing.com.",
-			refreshRate:      10 * time.Second,
-			ttl:              3 * time.Second,
-			expected:         []string{"10.0.0.2", "fd00::2"},
-			expectedAfterTTL: []string{"10.0.0.3", "fd00::3"},
-			registerDomain: func(domain string) {
-				fakeDNSServer.SetHosts(domain, 2)
-				fakeDNSServer.SetTTL(uint32(3))
-				time.AfterFunc(time.Second, func() {
-					fakeDNSServer.SetHosts(domain, 3)
-				})
-			},
-		},
-		{
-			name:             "check dns refresh after ttl without update bpfmap",
-			domain:           "www.test.com.",
-			refreshRate:      10 * time.Second,
-			ttl:              3 * time.Second,
-			expected:         []string{"10.0.0.2", "fd00::2"},
-			expectedAfterTTL: []string{"10.0.0.2", "fd00::2"},
-			registerDomain: func(domain string) {
-				fakeDNSServer.SetHosts(domain, 2)
-				fakeDNSServer.SetTTL(uint32(3))
-			},
-		},
-		{
-			name:             "check dns refresh after refreshRate, ttl > refreshRate",
-			domain:           "www.baidu.com.",
-			refreshRate:      3 * time.Second,
-			ttl:              10 * time.Second,
-			expected:         []string{"10.0.0.2", "fd00::2"},
-			expectedAfterTTL: []string{"10.0.0.3", "fd00::3"},
-			registerDomain: func(domain string) {
-				fakeDNSServer.SetHosts(domain, 2)
-				fakeDNSServer.SetTTL(uint32(10))
-				time.AfterFunc(time.Second, func() {
-					fakeDNSServer.SetHosts(domain, 3)
-				})
-			},
-		},
-		{
-			name:        "failed to resolve",
-			domain:      "www.kmesh.test.",
-			refreshRate: 10 * time.Second,
-			expected:    []string{},
-		},
-	}
-	var wg sync.WaitGroup
-	for _, testcase := range testCases {
-		wg.Add(1)
-		if testcase.registerDomain != nil {
-			testcase.registerDomain(testcase.domain)
-		}
-
-		input := &dns.PendingResolveDomain{
-			DomainName:  testcase.domain,
-			RefreshRate: testcase.refreshRate,
-		}
-		testDNSResolver.dnsResolver.Lock()
-		testDNSResolver.dnsResolver.Cache[testcase.domain] = &dns.DomainCacheEntry{}
-		testDNSResolver.dnsResolver.Unlock()
-		go testDNSResolver.refreshAdsWorker()
-
-		_, ttl, err := testDNSResolver.dnsResolver.Resolve(input.DomainName)
-		assert.NoError(t, err)
-		if ttl > input.RefreshRate {
-			ttl = input.RefreshRate
-		}
-		if ttl == 0 {
-			ttl = dns.DeRefreshInterval
-		}
-		testDNSResolver.dnsRefreshQueue.AddAfter(input, ttl)
-		time.Sleep(2 * time.Second)
-
-		res := testDNSResolver.dnsResolver.GetDNSAddresses(testcase.domain)
-		if len(res) != 0 || len(testcase.expected) != 0 {
-			if !reflect.DeepEqual(res, testcase.expected) {
-				t.Errorf("dns resolve for %s do not match. \n got %v\nwant %v", testcase.domain, res, testcase.expected)
-			}
-
-			if testcase.expectedAfterTTL != nil {
-				time.Sleep(ttl + 1)
-				res = testDNSResolver.dnsResolver.GetDNSAddresses(testcase.domain)
-				if !reflect.DeepEqual(res, testcase.expectedAfterTTL) {
-					t.Errorf("dns refresh after ttl failed, for %s do not match. \n got %v\nwant %v", testcase.domain, res, testcase.expectedAfterTTL)
-				}
-			}
-		}
-		wg.Done()
-	}
-	wg.Wait()
-}
-
 func TestGetPendingResolveDomain(t *testing.T) {
 	utCluster := clusterv3.Cluster{
 		Name: "testCluster",
@@ -411,7 +279,7 @@ func TestGetPendingResolveDomain(t *testing.T) {
 	tests := []struct {
 		name string
 		args args
-		want map[string]*dns.PendingResolveDomain
+		want map[string]*pendingResolveDomain
 	}{
 		{
 			name: "empty domains test",
@@ -420,7 +288,7 @@ func TestGetPendingResolveDomain(t *testing.T) {
 					&utCluster,
 				},
 			},
-			want: map[string]*dns.PendingResolveDomain{},
+			want: map[string]*pendingResolveDomain{},
 		},
 		{
 			name: "cluster domain is not IP",
@@ -429,7 +297,7 @@ func TestGetPendingResolveDomain(t *testing.T) {
 					&utClusterWithHost,
 				},
 			},
-			want: map[string]*dns.PendingResolveDomain{
+			want: map[string]*pendingResolveDomain{
 				"www.google.com": {
 					DomainName: "www.google.com",
 					Clusters:   []*clusterv3.Cluster{&utClusterWithHost},