status-im/consul
2
0
Fork 0
mirror of https://github.com/status-im/consul.git synced 2025-01-10 22:06:20 +00:00
Code Issues Projects Releases Wiki Activity

Check ingress upstreams when gating chain watches

Browse Source
This commit is contained in:
freddygv 2021-12-13 16:40:04 -07:00
parent 6814e84459
commit 2fe27b748d
3 changed files with 24 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
agent/proxycfg/ingress_gateway.go
View File

@ -128,6 +128,7 @@ func (s *handlerIngressGateway) handleUpdate(ctx context.Context, u cache.Update
} }
snap.IngressGateway.Upstreams = upstreamsMap snap.IngressGateway.Upstreams = upstreamsMap
snap.IngressGateway.UpstreamsSet = watchedSvcs
snap.IngressGateway.Hosts = hosts snap.IngressGateway.Hosts = hosts
snap.IngressGateway.HostsSet = true snap.IngressGateway.HostsSet = true

4
agent/proxycfg/snapshot.go
View File

@ -371,6 +371,9 @@ type configSnapshotIngressGateway struct {
// the GatewayServices RPC to retrieve them. // the GatewayServices RPC to retrieve them.
Upstreams map[IngressListenerKey]structs.Upstreams Upstreams map[IngressListenerKey]structs.Upstreams
// UpstreamsSet is the unique set of upstream.Identifier() the gateway routes to.
UpstreamsSet map[string]struct{}
// Listeners is the original listener config from the ingress-gateway config // Listeners is the original listener config from the ingress-gateway config
// entry to save us trying to pass fields through Upstreams // entry to save us trying to pass fields through Upstreams
Listeners map[IngressListenerKey]structs.IngressListener Listeners map[IngressListenerKey]structs.IngressListener
@ -381,6 +384,7 @@ func (c *configSnapshotIngressGateway) IsEmpty() bool {
return true return true
} }
return len(c.Upstreams) == 0 && return len(c.Upstreams) == 0 &&
len(c.UpstreamsSet) == 0 &&
len(c.DiscoveryChain) == 0 && len(c.DiscoveryChain) == 0 &&
len(c.WatchedUpstreams) == 0 && len(c.WatchedUpstreams) == 0 &&
len(c.WatchedUpstreamEndpoints) == 0 len(c.WatchedUpstreamEndpoints) == 0

24
agent/proxycfg/upstreams.go
View File

@ -44,11 +44,25 @@ func (s *handlerUpstreams) handleUpdateUpstreams(ctx context.Context, u cache.Up
} }
svc := strings.TrimPrefix(u.CorrelationID, "discovery-chain:") svc := strings.TrimPrefix(u.CorrelationID, "discovery-chain:")
explicit := snap.ConnectProxy.UpstreamConfig[svc].HasLocalPortOrSocket() switch snap.Kind {
if _, implicit := snap.ConnectProxy.IntentionUpstreams[svc]; !implicit && !explicit { case structs.ServiceKindIngressGateway:
// Discovery chain is not associated with a known explicit or implicit upstream so it is skipped. if _, ok := snap.IngressGateway.UpstreamsSet[svc]; !ok {
// The associated watch was likely cancelled. // Discovery chain is not associated with a known explicit or implicit upstream so it is skipped.
return nil // The associated watch was likely cancelled.
s.logger.Trace("discovery-chain watch fired for unknown upstream", "upstream", svc)
return nil
}
case structs.ServiceKindConnectProxy:
explicit := snap.ConnectProxy.UpstreamConfig[svc].HasLocalPortOrSocket()
if _, implicit := snap.ConnectProxy.IntentionUpstreams[svc]; !implicit && !explicit {
// Discovery chain is not associated with a known explicit or implicit upstream so it is skipped.
// The associated watch was likely cancelled.
s.logger.Trace("discovery-chain watch fired for unknown upstream", "upstream", svc)
return nil
}
default:
return fmt.Errorf("discovery-chain watch fired for unsupported kind: %s", snap.Kind)
} }
upstreamsSnapshot.DiscoveryChain[svc] = resp.Chain upstreamsSnapshot.DiscoveryChain[svc] = resp.Chain