From 0b376bda5be874a5c7052fce84a0fe76d9b630ec Mon Sep 17 00:00:00 2001 From: houyuwushang Date: Fri, 11 Sep 2026 11:55:05 +0800 Subject: [PATCH] test: cover Kubernetes autopath cache refresh (#8528) Exercise Pod-dependent search paths through the native Kubernetes plugin during prefetch and stale refresh, including dual-stack resolver results. Signed-off-by: houyuwushang --- test/kubernetes_autopath_cache_test.go | 289 +++++++++++++++++++++++++ 1 file changed, 289 insertions(+) create mode 100644 test/kubernetes_autopath_cache_test.go diff --git a/test/kubernetes_autopath_cache_test.go b/test/kubernetes_autopath_cache_test.go new file mode 100644 index 000000000..99fe21547 --- /dev/null +++ b/test/kubernetes_autopath_cache_test.go @@ -0,0 +1,289 @@ +package test + +import ( + "context" + "encoding/json" + "fmt" + "net" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "slices" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/coredns/coredns/plugin/pkg/dnstest" + + "github.com/miekg/dns" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/tools/clientcmd" + clientcmdapi "k8s.io/client-go/tools/clientcmd/api" +) + +// Background cache refreshes must not lose the Pod-dependent search path while +// refreshing an expanded external name. Keep AAAA fresh while A is refreshed to +// detect partial results from a dual-stack lookup as well as outright failures. +func TestKubernetesAutopathCacheRefresh(t *testing.T) { + for _, options := range []struct{ name, corefile string }{ + {"prefetch-stale", "prefetch 20\nserve_stale"}, + {"stale", "serve_stale"}, + } { + for _, network := range []string{"udp", "tcp"} { + t.Run(options.name+"/"+network, func(t *testing.T) { + // Kubernetes setup installs a process-global klog logger. Isolate + // each startup from other instances' informer goroutines. + const childEnv = "COREDNS_AUTOPATH_CACHE_TEST" + if os.Getenv(childEnv) != t.Name() { + pattern := "^" + strings.ReplaceAll(t.Name(), "/", "$/^") + "$" + cmd := exec.CommandContext(t.Context(), os.Args[0], "-test.run="+pattern, "-test.v", "-test.timeout=30s") + cmd.Env = append(os.Environ(), childEnv+"="+t.Name()) + if output, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("isolated test failed: %v\n%s", err, output) + } + return + } + kubeconfig := autopathKubeconfig(t) + const base = "external.example." + const expanded = base + "client.svc.cluster.local." + var aQueries atomic.Uint32 + upstream := dnstest.NewMultipleServer(func(w dns.ResponseWriter, r *dns.Msg) { + m := new(dns.Msg) + m.SetReply(r) + m.RecursionAvailable = true + q := r.Question[0] + h := dns.RR_Header{Name: q.Name, Rrtype: q.Qtype, Class: dns.ClassINET, Ttl: 300} + switch { + case q.Name == base && q.Qtype == dns.TypeA: + ip := net.ParseIP("192.0.2.2") + if aQueries.Add(1) == 1 { + h.Ttl = 1 + ip = net.ParseIP("192.0.2.1") + } + m.Answer = []dns.RR{&dns.A{Hdr: h, A: ip}} + case q.Name == base && q.Qtype == dns.TypeAAAA: + m.Answer = []dns.RR{&dns.AAAA{Hdr: h, AAAA: net.ParseIP("2001:db8::1")}} + default: + // Kubernetes also appends the host's resolv.conf search domains. + m.Rcode = dns.RcodeNameError + m.Ns = []dns.RR{&dns.SOA{ + Hdr: dns.RR_Header{Name: q.Name, Rrtype: dns.TypeSOA, Class: dns.ClassINET, Ttl: 30}, + Ns: "ns.example.", Mbox: "hostmaster.example.", Serial: 1, Minttl: 30, + }} + } + if err := w.WriteMsg(m); err != nil { + t.Error(err) + } + }) + t.Cleanup(upstream.Close) + corefile := fmt.Sprintf(`.:0 { + kubernetes cluster.local { + pods verified + kubeconfig %q + } + autopath @kubernetes + cache { + success 1024 3600 0 + denial 1024 1800 0 + %s + } + forward . %s + loadbalance +}`, filepath.ToSlash(kubeconfig), options.corefile, upstream.Addr) + server, udp, tcp, err := CoreDNSServerAndPorts(corefile) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + for _, err := range server.ShutdownCallbacks() { + t.Error(err) + } + if err := server.Stop(); err != nil { + t.Error(err) + } + server.Wait() + }) + addr := udp + if network == "tcp" { + addr = tcp + } + _, port, err := net.SplitHostPort(addr) + if err != nil { + t.Fatal(err) + } + addr = net.JoinHostPort("127.0.0.1", port) + query := func(name string, qtype uint16) *dns.Msg { + t.Helper() + m := new(dns.Msg) + m.SetQuestion(name, qtype) + m.SetEdns0(1232, false) + c := &dns.Client{Net: network, Timeout: 2 * time.Second} + r, _, err := c.Exchange(m, addr) + if err != nil { + t.Fatal(err) + } + return r + } + hasAddress := func(m *dns.Msg, ip string) bool { + return m.Rcode == dns.RcodeSuccess && slices.ContainsFunc(m.Answer, func(rr dns.RR) bool { + switch rr := rr.(type) { + case *dns.A: + return rr.A.String() == ip + case *dns.AAAA: + return rr.AAAA.String() == ip + } + return false + }) + } + // Verify that the real Kubernetes informers populated both indexes. + for name, ip := range map[string]string{ + "internal.client.svc.cluster.local.": "10.96.0.10", + "127-0-0-1.client.pod.cluster.local.": "127.0.0.1", + } { + if r := query(name, dns.TypeA); !hasAddress(r, ip) { + t.Fatalf("Kubernetes index check for %s failed: %s", name, r) + } + } + if r := query(expanded, dns.TypeAAAA); !hasAddress(r, "2001:db8::1") { + t.Fatalf("initial expanded AAAA lookup failed: %s", r) + } + if r := query(expanded, dns.TypeA); !hasAddress(r, "192.0.2.1") { + t.Fatalf("initial expanded A lookup failed: %s", r) + } + + // A changed address proves refresh completed. serve_stale ensures a + // delayed test still exercises background refresh, not a cache miss. + refreshed := false + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); { + r := query(expanded, dns.TypeA) + if hasAddress(r, "192.0.2.2") { + refreshed = true + break + } + if !hasAddress(r, "192.0.2.1") { + t.Errorf("expanded A lookup lost its answer during refresh: %s", r) + break + } + time.Sleep(10 * time.Millisecond) + } + if !refreshed { + t.Errorf("expanded A lookup did not receive the refreshed address; upstream A queries: %d", aQueries.Load()) + } + if r := query(base, dns.TypeA); !hasAddress(r, "192.0.2.2") { + t.Errorf("absolute A control lookup failed: %s", r) + } + if r := query(expanded, dns.TypeAAAA); !hasAddress(r, "2001:db8::1") { + t.Errorf("expanded AAAA control lookup failed: %s", r) + } + resolver := &net.Resolver{PreferGo: true, Dial: func(ctx context.Context, _, _ string) (net.Conn, error) { + d := net.Dialer{} + return d.DialContext(ctx, network, addr) + }} + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) + defer cancel() + ips, err := resolver.LookupIPAddr(ctx, expanded) + if err != nil || !slices.ContainsFunc(ips, func(ip net.IPAddr) bool { return ip.IP.String() == "192.0.2.2" }) || + !slices.ContainsFunc(ips, func(ip net.IPAddr) bool { return ip.IP.String() == "2001:db8::1" }) { + t.Errorf("expected both address families after refresh, got %v, err: %v", ips, err) + } + }) + } + } +} + +// Supply list/watch responses over HTTP, leaving the Kubernetes client, +// informers, Pod index and DNS/autopath handlers in the actual plugin chain. +func autopathKubeconfig(t *testing.T) string { + t.Helper() + resources := map[string]struct { + kind, apiVersion string + objects []runtime.Object + }{ + "/api/v1/namespaces": {"Namespace", "v1", []runtime.Object{&corev1.Namespace{ + TypeMeta: metav1.TypeMeta{Kind: "Namespace", APIVersion: "v1"}, + ObjectMeta: metav1.ObjectMeta{Name: "client", ResourceVersion: "1"}, + }}}, + "/api/v1/pods": {"Pod", "v1", []runtime.Object{&corev1.Pod{ + TypeMeta: metav1.TypeMeta{Kind: "Pod", APIVersion: "v1"}, + ObjectMeta: metav1.ObjectMeta{Name: "client", Namespace: "client", ResourceVersion: "1"}, + Status: corev1.PodStatus{PodIP: "127.0.0.1", PodIPs: []corev1.PodIP{{IP: "127.0.0.1"}}, Phase: corev1.PodRunning}, + }}}, + "/api/v1/services": {"Service", "v1", []runtime.Object{&corev1.Service{ + TypeMeta: metav1.TypeMeta{Kind: "Service", APIVersion: "v1"}, + ObjectMeta: metav1.ObjectMeta{Name: "internal", Namespace: "client", ResourceVersion: "1"}, + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeClusterIP, ClusterIP: "10.96.0.10", ClusterIPs: []string{"10.96.0.10"}, + Ports: []corev1.ServicePort{{Name: "http", Port: 80, Protocol: corev1.ProtocolTCP}}, + }, + }}}, + "/apis/discovery.k8s.io/v1/endpointslices": {"EndpointSlice", "discovery.k8s.io/v1", nil}, + } + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + resource, ok := resources[r.URL.Path] + if !ok || r.Method != http.MethodGet { + t.Errorf("unexpected Kubernetes API request: %s %s", r.Method, r.URL) + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "application/json") + enc := json.NewEncoder(w) + if r.URL.Query().Get("watch") != "true" { + items := make([]runtime.RawExtension, 0, len(resource.objects)) + for _, obj := range resource.objects { + items = append(items, runtime.RawExtension{Object: obj}) + } + if err := enc.Encode(&metav1.List{ + TypeMeta: metav1.TypeMeta{Kind: resource.kind + "List", APIVersion: resource.apiVersion}, + ListMeta: metav1.ListMeta{ResourceVersion: "1"}, Items: items, + }); err != nil { + t.Error(err) + } + return + } + // Support both list-then-watch and client-go's streaming initial list. + if r.URL.Query().Get("sendInitialEvents") == "true" { + send := func(eventType string, obj runtime.Object) { + if err := enc.Encode(struct { + Type string `json:"type"` + Object runtime.Object `json:"object"` + }{eventType, obj}); err != nil { + t.Error(err) + } + } + for _, obj := range resource.objects { + send("ADDED", obj) + } + send("BOOKMARK", &metav1.PartialObjectMetadata{ + TypeMeta: metav1.TypeMeta{Kind: resource.kind, APIVersion: resource.apiVersion}, + ObjectMeta: metav1.ObjectMeta{ResourceVersion: "1", Annotations: map[string]string{ + metav1.InitialEventsAnnotationKey: "true", + }}, + }) + } + w.(http.Flusher).Flush() + <-r.Context().Done() + })) + t.Cleanup(func() { + api.CloseClientConnections() + api.Close() + }) + config, err := clientcmd.Write(clientcmdapi.Config{ + Clusters: map[string]*clientcmdapi.Cluster{"test": {Server: api.URL}}, + AuthInfos: map[string]*clientcmdapi.AuthInfo{"test": {}}, + Contexts: map[string]*clientcmdapi.Context{"test": {Cluster: "test", AuthInfo: "test"}}, + CurrentContext: "test", + }) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "kubeconfig") + if err := os.WriteFile(path, config, 0600); err != nil { + t.Fatal(err) + } + return path +}