mirror of
https://github.com/coredns/coredns.git
synced 2026-10-08 19:45:21 -04:00
0b376bda5b
Exercise Pod-dependent search paths through the native Kubernetes plugin during prefetch and stale refresh, including dual-stack resolver results. Signed-off-by: houyuwushang <liuluoqianqiu@outlook.com>
290 lines
10 KiB
Go
290 lines
10 KiB
Go
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
|
|
}
|