luxolus
/
open-consul
2
0
Fork 0
Code Issues 1 Pull Requests Packages Releases Wiki Activity

xds: fix for delta xDS reconnect bug in LDS/CDS (#12174) 

When a wildcard xDS type (LDS/CDS/SRDS) reconnects from a delta xDS stream,
prior to envoy `1.19.0` it would populate the `ResourceNamesSubscribe` field
with the full list of currently subscribed items, instead of simply omitting it
to infer that it wanted everything (which is what wildcard mode means).

This upstream issue was filed in envoyproxy/envoy#16063 and fixed in
envoyproxy/envoy#16153 which went out in Envoy `1.19.0` and is fixed in later
versions (later refactored in envoyproxy/envoy#16855).

This PR conditionally forces LDS/CDS to be wildcard-only even when the
connected Envoy requests a non-wildcard subscription, but only does so on
versions prior to `1.19.0`, as we should not need to do this on later versions.

This fixes the failure case as described here: #11833 (comment)

Co-authored-by: Huan Wang <fredwanghuan@gmail.com>
This commit is contained in:
R.B. Boyer 2022-01-25 11:24:27 -06:00 committed by GitHub
parent ece75a03e3
commit b999b3edfc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 117 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.changelog/12174.txt Normal file
View File

@ -0,0 +1,3 @@
```release-note:bug
xds: fix for delta xDS reconnect bug in LDS/CDS
```

24
agent/xds/delta.go
View File

@ -165,12 +165,6 @@ func (s *Server) processDelta(stream ADSDeltaStream, reqCh <-chan *envoy_discove
return status.Errorf(codes.InvalidArgument, "type URL is required for ADS")
}
if handler, ok := handlers[req.TypeUrl]; ok {
if handler.Recv(req) {
generator.Logger.Trace("subscribing to type", "typeUrl", req.TypeUrl)
}
}
if node == nil && req.Node != nil {
node = req.Node
var err error
@ -180,6 +174,12 @@ func (s *Server) processDelta(stream ADSDeltaStream, reqCh <-chan *envoy_discove
}
}
if handler, ok := handlers[req.TypeUrl]; ok {
if handler.Recv(req, generator.ProxyFeatures) {
generator.Logger.Trace("subscribing to type", "typeUrl", req.TypeUrl)
}
}
case cfgSnap = <-stateCh:
newRes, err := generator.allResourcesFromSnapshot(cfgSnap)
if err != nil {
@ -434,7 +434,7 @@ func newDeltaType(
// Recv handles new discovery requests from envoy.
//
// Returns true the first time a type receives a request.
func (t *xDSDeltaType) Recv(req *envoy_discovery_v3.DeltaDiscoveryRequest) bool {
func (t *xDSDeltaType) Recv(req *envoy_discovery_v3.DeltaDiscoveryRequest, sf supportedProxyFeatures) bool {
if t == nil {
return false // not something we care about
}
@ -447,6 +447,16 @@ func (t *xDSDeltaType) Recv(req *envoy_discovery_v3.DeltaDiscoveryRequest) bool
t.wildcard = len(req.ResourceNamesSubscribe) == 0
t.registered = true
registeredThisTime = true
if sf.ForceLDSandCDSToAlwaysUseWildcardsOnReconnect {
switch t.typeURL {
case ListenerType, ClusterType:
if !t.wildcard {
t.wildcard = true
logger.Trace("fixing Envoy bug fixed in 1.19.0 by inferring wildcard mode for type")
}
}
}
}
/*

66
agent/xds/delta_test.go
View File

@ -554,6 +554,72 @@ func TestServer_DeltaAggregatedResources_v3_SlowEndpointPopulation(t *testing.T)
}
}
func TestServer_DeltaAggregatedResources_v3_GetAllClusterAfterConsulRestarted(t *testing.T) {
// This illustrates a scenario related to https://github.com/hashicorp/consul/issues/11833
aclResolve := func(id string) (acl.Authorizer, error) {
// Allow all
return acl.RootAuthorizer("manage"), nil
}
scenario := newTestServerDeltaScenario(t, aclResolve, "web-sidecar-proxy", "", 0)
_, mgr, errCh, envoy := scenario.server, scenario.mgr, scenario.errCh, scenario.envoy
envoy.EnvoyVersion = "1.18.0"
sid := structs.NewServiceID("web-sidecar-proxy", nil)
// Register the proxy to create state needed to Watch() on
mgr.RegisterProxy(t, sid)
var snap *proxycfg.ConfigSnapshot
runStep(t, "get into state after consul restarted", func(t *testing.T) {
snap = newTestSnapshot(t, nil, "")
// Send initial cluster discover.
// This is to simulate the discovery request call from envoy after disconnected from consul ads stream.
//
// We need to force it to be an older version of envoy so that the logic shifts.
envoy.SendDeltaReq(t, ClusterType, &envoy_discovery_v3.DeltaDiscoveryRequest{
ResourceNamesSubscribe: []string{
"local_app",
"db.default.dc1.internal.11111111-2222-3333-4444-555555555555.consul",
},
InitialResourceVersions: map[string]string{
"local_app": "a948904f2f0f479b8f8197694b30184b0d2ed1c1cd2a1ec0fb85d299a192a447",
"db.default.dc1.internal.11111111-2222-3333-4444-555555555555.consul": "5891b5b522d5df086d0ff0b110fbd9d21bb4fc7163af34d08286a2e846f6be03",
},
})
// Check no response sent yet
assertDeltaChanBlocked(t, envoy.deltaStream.sendCh)
requireProtocolVersionGauge(t, scenario, "v3", 1)
// Deliver a new snapshot
// the config contains 3 clusters: local_app, db, geo-cache.
// this is to simulate the fact that there is one additional (upstream) cluster gets added to the sidecar service
// during the time xds disconnected (consul restarted).
mgr.DeliverConfig(t, sid, snap)
assertDeltaResponseSent(t, envoy.deltaStream.sendCh, &envoy_discovery_v3.DeltaDiscoveryResponse{
TypeUrl: ClusterType,
Nonce: hexString(1),
Resources: makeTestResources(t,
makeTestCluster(t, snap, "tcp:local_app"),
makeTestCluster(t, snap, "tcp:db"),
makeTestCluster(t, snap, "tcp:geo-cache"),
),
})
})
envoy.Close()
select {
case err := <-errCh:
require.NoError(t, err)
case <-time.After(50 * time.Millisecond):
t.Fatalf("timed out waiting for handler to finish")
}
}
func TestServer_DeltaAggregatedResources_v3_BasicProtocol_TCP_clusterChangesImpactEndpoints(t *testing.T) {
aclResolve := func(id string) (acl.Authorizer, error) {
// Allow all

31
agent/xds/envoy_versioning.go
View File

@ -13,8 +13,7 @@ var (
// the zero'th point release of the last element of proxysupport.EnvoyVersions.
minSupportedVersion = version.Must(version.NewVersion("1.17.0"))
// add min version constraints for associated feature flags when necessary, for example:
// minVersionAllowingEmptyGatewayClustersWithIncrementalXDS = version.Must(version.NewVersion("1.16.0"))
minVersionToForceLDSandCDSToAlwaysUseWildcardsOnReconnect = version.Must(version.NewVersion("1.19.0"))
specificUnsupportedVersions = []unsupportedVersion{}
)
@ -26,16 +25,19 @@ type unsupportedVersion struct {
}
type supportedProxyFeatures struct {
// add version dependent feature flags here
// Older versions of Envoy incorrectly exploded a wildcard subscription for
// LDS and CDS into specific line items on incremental xDS reconnect. They
// would populate both InitialResourceVersions and ResourceNamesSubscribe
// when they SHOULD have left ResourceNamesSubscribe empty (or used an
// explicit "*" in later Envoy versions) to imply wildcard mode. On
// reconnect, Consul interpreted the lack of the wildcard attribute as
// implying that the Envoy instance should not receive updates for any
// newly created listeners and clusters for the remaining life of that
// Envoy sidecar process.
//
// For example, we previously had flags for Envoy < 1.16 called:
//
// GatewaysNeedStubClusterWhenEmptyWithIncrementalXDS
// IncrementalXDSUpdatesMustBeSerial
//
// Which then manifested in the code for checks with this struct populated.
// By dropping support for 1.15, we no longer have any special flags here
// but leaving this flagging functionality for future one-offs.
// see: https://github.com/envoyproxy/envoy/issues/16063
// see: https://github.com/envoyproxy/envoy/pull/16153
ForceLDSandCDSToAlwaysUseWildcardsOnReconnect bool
}
func determineSupportedProxyFeatures(node *envoy_core_v3.Node) (supportedProxyFeatures, error) {
@ -73,12 +75,9 @@ func determineSupportedProxyFeaturesFromVersion(version *version.Version) (suppo
sf := supportedProxyFeatures{}
// add version constraints to populate feature flags here when necessary, for example:
/*
if version.LessThan(minVersionAllowingEmptyGatewayClustersWithIncrementalXDS) {
sf.GatewaysNeedStubClusterWhenEmptyWithIncrementalXDS = true
if version.LessThan(minVersionToForceLDSandCDSToAlwaysUseWildcardsOnReconnect) {
sf.ForceLDSandCDSToAlwaysUseWildcardsOnReconnect = true
}
*/
return sf, nil
}

6
agent/xds/envoy_versioning_test.go
View File

@ -125,6 +125,12 @@ func TestDetermineSupportedProxyFeaturesFromString(t *testing.T) {
for _, v := range []string{
"1.17.0", "1.17.1", "1.17.2", "1.17.3", "1.17.4",
"1.18.0", "1.18.1", "1.18.2", "1.18.3", "1.18.4",
} {
cases[v] = testcase{expect: supportedProxyFeatures{
ForceLDSandCDSToAlwaysUseWildcardsOnReconnect: true,
}}
}
for _, v := range []string{
"1.19.0", "1.19.1",
"1.20.0", "1.20.1",
} {

11
agent/xds/testing.go
View File

@ -83,6 +83,8 @@ type TestEnvoy struct {
proxyID string
token string
EnvoyVersion string
deltaStream *TestADSDeltaStream // Incremental v3
}
@ -182,9 +184,14 @@ func (e *TestEnvoy) sendDeltaReq(
e.mu.Lock()
defer e.mu.Unlock()
ev, valid := stringToEnvoyVersion(proxysupport.EnvoyVersions[0])
stringVersion := e.EnvoyVersion
if stringVersion == "" {
stringVersion = proxysupport.EnvoyVersions[0]
}
ev, valid := stringToEnvoyVersion(stringVersion)
if !valid {
t.Fatal("envoy version is not valid: %s", proxysupport.EnvoyVersions[0])
t.Fatal("envoy version is not valid: %s", stringVersion)
}
if req == nil {