diff --git a/CHANGELOG.md b/CHANGELOG.md index 6cc3578201..1c8ef1acbd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -84,6 +84,7 @@ * [BUGFIX] Ring: Fix DynamoDB KV CAS not retrying on transactional conditional check failures. `TransactWriteItems` reports condition failures as `TransactionCanceledException` with a `ConditionalCheckFailed` cancellation reason, which was not recognized as retryable, so any concurrent ring update conflict (e.g. many ingesters joining during a rolling update) failed immediately instead of re-reading and retrying. `TransactionConflict` cancellation reasons are also treated as retryable. #7706 * [BUGFIX] Distributor: Return HTTP 499 (Client Closed Request) instead of 500 when a remote-write or OTLP push is canceled by the client, so client-side cancellations are no longer counted as server-side errors. #7717 * [BUGFIX] Querier: Fix gRPC `codes.Canceled` errors being mapped to HTTP 500 instead of 499 when a client cancels a query. #7738 +* [BUGFIX] Fix the gRPC DNS watcher's SRV record path deleting all known endpoints when the SRV query succeeds but every target's A record lookup fails. #7745 ## 1.21.1 2026-06-04 diff --git a/pkg/util/grpcutil/dns_resolver.go b/pkg/util/grpcutil/dns_resolver.go index b99f0a0e98..0e2e00acb1 100644 --- a/pkg/util/grpcutil/dns_resolver.go +++ b/pkg/util/grpcutil/dns_resolver.go @@ -188,6 +188,10 @@ func (w *dnsWatcher) compileUpdate(newAddrs map[string]*Update) []*Update { return res } +// lookupSRV resolves the SRV record of the watched service and then the A +// records of every SRV target. It returns nil when the lookup failed; a +// non-nil empty map is an authoritative result meaning the service has no +// endpoints. func (w *dnsWatcher) lookupSRV() map[string]*Update { if w.service == "" { return nil @@ -199,7 +203,15 @@ func (w *dnsWatcher) lookupSRV() map[string]*Update { level.Info(w.logger).Log("msg", "failed DNS SRV record lookup", "err", err) return nil } + resolvableTargets := 0 for _, s := range srvs { + if s.Target == "." { + // RFC 2782: a target of "." means the service is decidedly not + // available at this domain. It contributes no addresses, but it + // is an authoritative answer, not a lookup failure. + continue + } + resolvableTargets++ addrs, err := lookupHost(w.ctx, s.Target) if err != nil { level.Warn(w.logger).Log("msg", "failed SRV target DNS lookup", "target", s.Target, "err", err) @@ -208,16 +220,32 @@ func (w *dnsWatcher) lookupSRV() map[string]*Update { for _, a := range addrs { a, ok := formatIP(a) if !ok { - level.Error(w.logger).Log("failed IP parsing", "err", err) + level.Error(w.logger).Log("msg", "failed IP parsing", "err", err) continue } addr := a + ":" + strconv.Itoa(int(s.Port)) newAddrs[addr] = &Update{Addr: addr} } } + if resolvableTargets > 0 && len(newAddrs) == 0 { + // The SRV query returned resolvable targets, but none of them yielded + // a usable address. Return nil to signal a failed lookup, honouring + // the same nil-means-failure contract as lookupHost(), so that + // lookup() falls back to the plain A record lookup of the watched + // host (note: that fallback uses the watcher's port, which may not + // match the SRV-advertised ports) and, failing that too, retains the + // previously resolved addresses instead of deleting them all. A + // non-nil empty map remains reserved for an authoritative empty + // answer — an SRV query returning zero records or only "." targets — + // which still propagates as a deletion of all known addresses. + return nil + } return newAddrs } +// lookupHost resolves the A records of the watched host. It returns nil when +// the lookup failed; a non-nil empty map is an authoritative result meaning +// the host has no addresses. func (w *dnsWatcher) lookupHost() map[string]*Update { newAddrs := make(map[string]*Update) addrs, err := lookupHost(w.ctx, w.host) diff --git a/pkg/util/grpcutil/dns_resolver_test.go b/pkg/util/grpcutil/dns_resolver_test.go index 9098c067eb..8bd60a5781 100644 --- a/pkg/util/grpcutil/dns_resolver_test.go +++ b/pkg/util/grpcutil/dns_resolver_test.go @@ -3,6 +3,7 @@ package grpcutil import ( "context" "errors" + "net" "testing" "time" @@ -18,6 +19,13 @@ func stubLookups(t *testing.T, host func(ctx context.Context, host string) ([]st t.Cleanup(func() { lookupHost = orig }) } +func stubSRVLookup(t *testing.T, srv func(ctx context.Context, service, proto, name string) (string, []*net.SRV, error)) { + t.Helper() + orig := lookupSRV + lookupSRV = srv + t.Cleanup(func() { lookupSRV = orig }) +} + func newTestDNSWatcher() *dnsWatcher { ctx, cancel := context.WithCancel(context.Background()) return &dnsWatcher{ @@ -78,3 +86,180 @@ func TestDNSWatcher_Lookup_TransientFailureRetainsCache(t *testing.T) { assert.Nil(t, result) assert.Equal(t, map[string]*Update{"1.2.3.4:80": {Addr: "1.2.3.4:80"}}, w.curAddrs) } + +func TestDNSWatcher_Lookup_SRVTargetTransientFailureRetainsCache(t *testing.T) { + stubSRVLookup(t, func(context.Context, string, string, string) (string, []*net.SRV, error) { + return "", []*net.SRV{ + {Target: "target-1.example.com", Port: 9095}, + {Target: "target-2.example.com", Port: 9095}, + }, nil + }) + stubLookups(t, func(context.Context, string) ([]string, error) { + return nil, errors.New("unable to resolve address") + }) + + w := newTestDNSWatcher() + w.service = "grpc" + w.curAddrs = map[string]*Update{"1.2.3.4:9095": {Addr: "1.2.3.4:9095"}} + + result := w.lookup() + + assert.Nil(t, result) + assert.Equal(t, map[string]*Update{"1.2.3.4:9095": {Addr: "1.2.3.4:9095"}}, w.curAddrs) +} + +func TestDNSWatcher_Lookup_SRVPartialTargetSuccessUpdatesCache(t *testing.T) { + stubSRVLookup(t, func(context.Context, string, string, string) (string, []*net.SRV, error) { + return "", []*net.SRV{ + {Target: "target-1.example.com", Port: 9095}, + {Target: "target-2.example.com", Port: 9095}, + }, nil + }) + stubLookups(t, func(_ context.Context, host string) ([]string, error) { + switch host { + case "target-1.example.com": + return []string{"5.6.7.8"}, nil + case "target-2.example.com": + return nil, errors.New("unable to resolve address") + default: + t.Fatalf("unexpected host lookup: %s", host) + return nil, nil + } + }) + + w := newTestDNSWatcher() + w.service = "grpc" + w.curAddrs = map[string]*Update{"1.2.3.4:9095": {Addr: "1.2.3.4:9095"}} + + result := w.lookup() + + assert.ElementsMatch(t, []*Update{ + {Op: Delete, Addr: "1.2.3.4:9095"}, + {Op: Add, Addr: "5.6.7.8:9095"}, + }, result) + assert.Equal(t, map[string]*Update{"5.6.7.8:9095": {Addr: "5.6.7.8:9095"}}, w.curAddrs) +} + +func TestDNSWatcher_Lookup_EmptySRVResultClearsCache(t *testing.T) { + stubSRVLookup(t, func(context.Context, string, string, string) (string, []*net.SRV, error) { + return "", []*net.SRV{}, nil + }) + stubLookups(t, func(_ context.Context, host string) ([]string, error) { + t.Fatalf("unexpected host lookup: %s", host) + return nil, nil + }) + + w := newTestDNSWatcher() + w.service = "grpc" + w.curAddrs = map[string]*Update{"1.2.3.4:9095": {Addr: "1.2.3.4:9095"}} + + result := w.lookup() + + assert.Equal(t, []*Update{{Op: Delete, Addr: "1.2.3.4:9095"}}, result) + assert.NotNil(t, w.curAddrs) + assert.Empty(t, w.curAddrs) +} + +func TestDNSWatcher_Lookup_SRVFailureFallsBackToHost(t *testing.T) { + stubSRVLookup(t, func(context.Context, string, string, string) (string, []*net.SRV, error) { + return "", nil, errors.New("unable to resolve SRV record") + }) + stubLookups(t, func(_ context.Context, host string) ([]string, error) { + assert.Equal(t, "myhost", host) + return []string{"5.6.7.8"}, nil + }) + + w := newTestDNSWatcher() + w.service = "grpc" + + result := w.lookup() + + assert.Equal(t, []*Update{{Op: Add, Addr: "5.6.7.8:80"}}, result) + assert.Equal(t, map[string]*Update{"5.6.7.8:80": {Addr: "5.6.7.8:80"}}, w.curAddrs) +} + +func TestDNSWatcher_Lookup_SRVAllTargetsFailFallsBackToHost(t *testing.T) { + stubSRVLookup(t, func(context.Context, string, string, string) (string, []*net.SRV, error) { + return "", []*net.SRV{ + {Target: "target-1.example.com", Port: 9095}, + }, nil + }) + stubLookups(t, func(_ context.Context, host string) ([]string, error) { + // The SRV target lookup fails, but the plain A record lookup on the + // watcher's host succeeds: the resolved addresses must then come from + // the A record fallback, using the watcher's port. + if host == "target-1.example.com" { + return nil, errors.New("unable to resolve address") + } + assert.Equal(t, "myhost", host) + return []string{"5.6.7.8"}, nil + }) + + w := newTestDNSWatcher() + w.service = "grpc" + w.curAddrs = map[string]*Update{"1.2.3.4:9095": {Addr: "1.2.3.4:9095"}} + + result := w.lookup() + + assert.ElementsMatch(t, []*Update{ + {Op: Delete, Addr: "1.2.3.4:9095"}, + {Op: Add, Addr: "5.6.7.8:80"}, + }, result) + assert.Equal(t, map[string]*Update{"5.6.7.8:80": {Addr: "5.6.7.8:80"}}, w.curAddrs) +} + +func TestDNSWatcher_Lookup_SRVTargetDotClearsCache(t *testing.T) { + stubSRVLookup(t, func(context.Context, string, string, string) (string, []*net.SRV, error) { + // RFC 2782: a single SRV record with target "." means the service is + // decidedly not available at this domain. + return "", []*net.SRV{{Target: ".", Port: 9095}}, nil + }) + stubLookups(t, func(_ context.Context, host string) ([]string, error) { + t.Fatalf("unexpected host lookup: %s", host) + return nil, nil + }) + + w := newTestDNSWatcher() + w.service = "grpc" + w.curAddrs = map[string]*Update{"1.2.3.4:9095": {Addr: "1.2.3.4:9095"}} + + result := w.lookup() + + assert.Equal(t, []*Update{{Op: Delete, Addr: "1.2.3.4:9095"}}, result) + assert.Empty(t, w.curAddrs) +} + +func TestDNSWatcher_Lookup_SRVTargetFailureAcrossPolls(t *testing.T) { + stubSRVLookup(t, func(context.Context, string, string, string) (string, []*net.SRV, error) { + return "", []*net.SRV{{Target: "target-1.example.com", Port: 9095}}, nil + }) + targetAddrs := []string{"1.2.3.4"} + targetErr := error(nil) + stubLookups(t, func(_ context.Context, host string) ([]string, error) { + if host == "target-1.example.com" { + return targetAddrs, targetErr + } + // The plain host A record fallback also fails during the outage. + return nil, errors.New("unable to resolve address") + }) + + w := newTestDNSWatcher() + w.service = "grpc" + + // The first poll succeeds and seeds the cache. + result := w.lookup() + assert.ElementsMatch(t, []*Update{{Op: Add, Addr: "1.2.3.4:9095"}}, result) + + // The target keeps failing for two consecutive polls: no updates are + // emitted and the cached endpoint is retained. + targetAddrs, targetErr = nil, errors.New("unable to resolve address") + for range 2 { + assert.Nil(t, w.lookup()) + assert.Equal(t, map[string]*Update{"1.2.3.4:9095": {Addr: "1.2.3.4:9095"}}, w.curAddrs) + } + + // Resolution recovers with the same address: no churn is emitted. + targetAddrs, targetErr = []string{"1.2.3.4"}, nil + assert.Empty(t, w.lookup()) + assert.Equal(t, map[string]*Update{"1.2.3.4:9095": {Addr: "1.2.3.4:9095"}}, w.curAddrs) +}