8eaca35df1
Transparent proxies typically cannot dial upstreams in remote datacenters. However, if their upstream configures a redirect to a remote DC then the upstream targets will be in another datacenter. In that sort of case we should use the WAN address for the passthrough.
446 lines
14 KiB
Go
446 lines
14 KiB
Go
package proxycfg
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/mitchellh/mapstructure"
|
|
|
|
"github.com/hashicorp/consul/agent/cache"
|
|
cachetype "github.com/hashicorp/consul/agent/cache-types"
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
)
|
|
|
|
type handlerUpstreams struct {
|
|
handlerState
|
|
}
|
|
|
|
func (s *handlerUpstreams) handleUpdateUpstreams(ctx context.Context, u cache.UpdateEvent, snap *ConfigSnapshot) error {
|
|
if u.Err != nil {
|
|
return fmt.Errorf("error filling agent cache: %v", u.Err)
|
|
}
|
|
|
|
upstreamsSnapshot := &snap.ConnectProxy.ConfigSnapshotUpstreams
|
|
if snap.Kind == structs.ServiceKindIngressGateway {
|
|
upstreamsSnapshot = &snap.IngressGateway.ConfigSnapshotUpstreams
|
|
}
|
|
|
|
switch {
|
|
case u.CorrelationID == leafWatchID:
|
|
leaf, ok := u.Result.(*structs.IssuedCert)
|
|
if !ok {
|
|
return fmt.Errorf("invalid type for response: %T", u.Result)
|
|
}
|
|
upstreamsSnapshot.Leaf = leaf
|
|
|
|
case strings.HasPrefix(u.CorrelationID, "discovery-chain:"):
|
|
resp, ok := u.Result.(*structs.DiscoveryChainResponse)
|
|
if !ok {
|
|
return fmt.Errorf("invalid type for response: %T", u.Result)
|
|
}
|
|
uidString := strings.TrimPrefix(u.CorrelationID, "discovery-chain:")
|
|
uid := UpstreamIDFromString(uidString)
|
|
|
|
switch snap.Kind {
|
|
case structs.ServiceKindIngressGateway:
|
|
if _, ok := snap.IngressGateway.UpstreamsSet[uid]; !ok {
|
|
// Discovery chain is not associated with a known explicit or implicit upstream so it is purged/skipped.
|
|
// The associated watch was likely cancelled.
|
|
delete(upstreamsSnapshot.DiscoveryChain, uid)
|
|
s.logger.Trace("discovery-chain watch fired for unknown upstream", "upstream", uid)
|
|
return nil
|
|
}
|
|
|
|
case structs.ServiceKindConnectProxy:
|
|
explicit := snap.ConnectProxy.UpstreamConfig[uid].HasLocalPortOrSocket()
|
|
if _, implicit := snap.ConnectProxy.IntentionUpstreams[uid]; !implicit && !explicit {
|
|
// Discovery chain is not associated with a known explicit or implicit upstream so it is purged/skipped.
|
|
// The associated watch was likely cancelled.
|
|
delete(upstreamsSnapshot.DiscoveryChain, uid)
|
|
s.logger.Trace("discovery-chain watch fired for unknown upstream", "upstream", uid)
|
|
return nil
|
|
}
|
|
default:
|
|
return fmt.Errorf("discovery-chain watch fired for unsupported kind: %s", snap.Kind)
|
|
}
|
|
|
|
upstreamsSnapshot.DiscoveryChain[uid] = resp.Chain
|
|
|
|
if err := s.resetWatchesFromChain(ctx, uid, resp.Chain, upstreamsSnapshot); err != nil {
|
|
return err
|
|
}
|
|
|
|
case strings.HasPrefix(u.CorrelationID, "upstream-target:"):
|
|
resp, ok := u.Result.(*structs.IndexedCheckServiceNodes)
|
|
if !ok {
|
|
return fmt.Errorf("invalid type for response: %T", u.Result)
|
|
}
|
|
correlationID := strings.TrimPrefix(u.CorrelationID, "upstream-target:")
|
|
targetID, uidString, ok := removeColonPrefix(correlationID)
|
|
if !ok {
|
|
return fmt.Errorf("invalid correlation id %q", u.CorrelationID)
|
|
}
|
|
|
|
uid := UpstreamIDFromString(uidString)
|
|
|
|
if _, ok := upstreamsSnapshot.WatchedUpstreamEndpoints[uid]; !ok {
|
|
upstreamsSnapshot.WatchedUpstreamEndpoints[uid] = make(map[string]structs.CheckServiceNodes)
|
|
}
|
|
upstreamsSnapshot.WatchedUpstreamEndpoints[uid][targetID] = resp.Nodes
|
|
|
|
if s.kind != structs.ServiceKindConnectProxy || s.proxyCfg.Mode != structs.ProxyModeTransparent {
|
|
return nil
|
|
}
|
|
|
|
// Clear out this target's existing passthrough upstreams and indices so that they can be repopulated below.
|
|
if _, ok := upstreamsSnapshot.PassthroughUpstreams[uid]; ok {
|
|
for addr := range upstreamsSnapshot.PassthroughUpstreams[uid][targetID] {
|
|
if indexed := upstreamsSnapshot.PassthroughIndices[addr]; indexed.targetID == targetID && indexed.upstreamID == uid {
|
|
delete(upstreamsSnapshot.PassthroughIndices, addr)
|
|
}
|
|
}
|
|
upstreamsSnapshot.PassthroughUpstreams[uid][targetID] = make(map[string]struct{})
|
|
}
|
|
|
|
passthroughs := make(map[string]struct{})
|
|
|
|
for _, node := range resp.Nodes {
|
|
if !node.Service.Proxy.TransparentProxy.DialedDirectly {
|
|
continue
|
|
}
|
|
|
|
// Make sure to use an external address when crossing partition or DC boundaries.
|
|
isRemote := !snap.Locality.Matches(node.Node.Datacenter, node.Node.PartitionOrDefault())
|
|
csnIdx, addr, _ := node.BestAddress(isRemote)
|
|
|
|
existing := upstreamsSnapshot.PassthroughIndices[addr]
|
|
if existing.idx > csnIdx {
|
|
// The last known instance with this address had a higher index so it takes precedence.
|
|
continue
|
|
}
|
|
|
|
// The current instance has a higher Raft index so we ensure the passthrough address is only
|
|
// associated with this upstream target. Older associations are cleaned up as needed.
|
|
delete(upstreamsSnapshot.PassthroughUpstreams[existing.upstreamID][existing.targetID], addr)
|
|
if len(upstreamsSnapshot.PassthroughUpstreams[existing.upstreamID][existing.targetID]) == 0 {
|
|
delete(upstreamsSnapshot.PassthroughUpstreams[existing.upstreamID], existing.targetID)
|
|
}
|
|
if len(upstreamsSnapshot.PassthroughUpstreams[existing.upstreamID]) == 0 {
|
|
delete(upstreamsSnapshot.PassthroughUpstreams, existing.upstreamID)
|
|
}
|
|
|
|
upstreamsSnapshot.PassthroughIndices[addr] = indexedTarget{idx: csnIdx, upstreamID: uid, targetID: targetID}
|
|
passthroughs[addr] = struct{}{}
|
|
}
|
|
if len(passthroughs) > 0 {
|
|
upstreamsSnapshot.PassthroughUpstreams[uid] = map[string]map[string]struct{}{
|
|
targetID: passthroughs,
|
|
}
|
|
}
|
|
|
|
case strings.HasPrefix(u.CorrelationID, "mesh-gateway:"):
|
|
resp, ok := u.Result.(*structs.IndexedNodesWithGateways)
|
|
if !ok {
|
|
return fmt.Errorf("invalid type for response: %T", u.Result)
|
|
}
|
|
correlationID := strings.TrimPrefix(u.CorrelationID, "mesh-gateway:")
|
|
key, uidString, ok := removeColonPrefix(correlationID)
|
|
if !ok {
|
|
return fmt.Errorf("invalid correlation id %q", u.CorrelationID)
|
|
}
|
|
uid := UpstreamIDFromString(uidString)
|
|
|
|
if _, ok = upstreamsSnapshot.WatchedGatewayEndpoints[uid]; !ok {
|
|
upstreamsSnapshot.WatchedGatewayEndpoints[uid] = make(map[string]structs.CheckServiceNodes)
|
|
}
|
|
upstreamsSnapshot.WatchedGatewayEndpoints[uid][key] = resp.Nodes
|
|
|
|
default:
|
|
return fmt.Errorf("unknown correlation ID: %s", u.CorrelationID)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func removeColonPrefix(s string) (string, string, bool) {
|
|
idx := strings.Index(s, ":")
|
|
if idx == -1 {
|
|
return "", "", false
|
|
}
|
|
return s[0:idx], s[idx+1:], true
|
|
}
|
|
|
|
func (s *handlerUpstreams) resetWatchesFromChain(
|
|
ctx context.Context,
|
|
uid UpstreamID,
|
|
chain *structs.CompiledDiscoveryChain,
|
|
snap *ConfigSnapshotUpstreams,
|
|
) error {
|
|
s.logger.Trace("resetting watches for discovery chain", "id", uid)
|
|
if chain == nil {
|
|
return fmt.Errorf("not possible to arrive here with no discovery chain")
|
|
}
|
|
|
|
// Initialize relevant sub maps.
|
|
if _, ok := snap.WatchedUpstreams[uid]; !ok {
|
|
snap.WatchedUpstreams[uid] = make(map[string]context.CancelFunc)
|
|
}
|
|
if _, ok := snap.WatchedUpstreamEndpoints[uid]; !ok {
|
|
snap.WatchedUpstreamEndpoints[uid] = make(map[string]structs.CheckServiceNodes)
|
|
}
|
|
if _, ok := snap.WatchedGateways[uid]; !ok {
|
|
snap.WatchedGateways[uid] = make(map[string]context.CancelFunc)
|
|
}
|
|
if _, ok := snap.WatchedGatewayEndpoints[uid]; !ok {
|
|
snap.WatchedGatewayEndpoints[uid] = make(map[string]structs.CheckServiceNodes)
|
|
}
|
|
|
|
// We could invalidate this selectively based on a hash of the relevant
|
|
// resolver information, but for now just reset anything about this
|
|
// upstream when the chain changes in any way.
|
|
//
|
|
// TODO(rb): content hash based add/remove
|
|
for targetID, cancelFn := range snap.WatchedUpstreams[uid] {
|
|
s.logger.Trace("stopping watch of target",
|
|
"upstream", uid,
|
|
"chain", chain.ServiceName,
|
|
"target", targetID,
|
|
)
|
|
delete(snap.WatchedUpstreams[uid], targetID)
|
|
delete(snap.WatchedUpstreamEndpoints[uid], targetID)
|
|
cancelFn()
|
|
}
|
|
|
|
var (
|
|
watchedChainEndpoints bool
|
|
needGateways = make(map[string]struct{})
|
|
)
|
|
|
|
chainID := chain.ID()
|
|
for _, target := range chain.Targets {
|
|
if target.ID == chainID {
|
|
watchedChainEndpoints = true
|
|
}
|
|
|
|
opts := targetWatchOpts{
|
|
upstreamID: uid,
|
|
chainID: target.ID,
|
|
service: target.Service,
|
|
filter: target.Subset.Filter,
|
|
datacenter: target.Datacenter,
|
|
entMeta: target.GetEnterpriseMetadata(),
|
|
}
|
|
err := s.watchUpstreamTarget(ctx, snap, opts)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to watch target %q for upstream %q", target.ID, uid)
|
|
}
|
|
|
|
// We'll get endpoints from the gateway query, but the health still has
|
|
// to come from the backing service query.
|
|
var gk GatewayKey
|
|
|
|
switch target.MeshGateway.Mode {
|
|
case structs.MeshGatewayModeRemote:
|
|
gk = GatewayKey{
|
|
Partition: target.Partition,
|
|
Datacenter: target.Datacenter,
|
|
}
|
|
case structs.MeshGatewayModeLocal:
|
|
gk = GatewayKey{
|
|
Partition: s.source.NodePartitionOrDefault(),
|
|
Datacenter: s.source.Datacenter,
|
|
}
|
|
}
|
|
if s.source.Datacenter != target.Datacenter || s.proxyID.PartitionOrDefault() != target.Partition {
|
|
needGateways[gk.String()] = struct{}{}
|
|
}
|
|
}
|
|
|
|
// If the discovery chain's targets do not lead to watching all endpoints
|
|
// for the upstream, then create a separate watch for those too.
|
|
// This is needed in transparent mode because if there is some service A that
|
|
// redirects to service B, the dialing proxy needs to associate A's virtual IP
|
|
// with A's discovery chain.
|
|
//
|
|
// Outside of transparent mode we only watch the chain target, B,
|
|
// since A is a virtual service and traffic will not be sent to it.
|
|
if !watchedChainEndpoints && s.proxyCfg.Mode == structs.ProxyModeTransparent {
|
|
chainEntMeta := structs.NewEnterpriseMetaWithPartition(chain.Partition, chain.Namespace)
|
|
|
|
opts := targetWatchOpts{
|
|
upstreamID: uid,
|
|
chainID: chainID,
|
|
service: chain.ServiceName,
|
|
filter: "",
|
|
datacenter: chain.Datacenter,
|
|
entMeta: &chainEntMeta,
|
|
}
|
|
err := s.watchUpstreamTarget(ctx, snap, opts)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to watch target %q for upstream %q", chainID, uid)
|
|
}
|
|
}
|
|
|
|
for key := range needGateways {
|
|
if _, ok := snap.WatchedGateways[uid][key]; ok {
|
|
continue
|
|
}
|
|
gwKey := gatewayKeyFromString(key)
|
|
|
|
s.logger.Trace("initializing watch of mesh gateway",
|
|
"upstream", uid,
|
|
"chain", chain.ServiceName,
|
|
"datacenter", gwKey.Datacenter,
|
|
"partition", gwKey.Partition,
|
|
)
|
|
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
opts := gatewayWatchOpts{
|
|
notifier: s.cache,
|
|
notifyCh: s.ch,
|
|
source: *s.source,
|
|
token: s.token,
|
|
key: gwKey,
|
|
upstreamID: uid,
|
|
}
|
|
err := watchMeshGateway(ctx, opts)
|
|
if err != nil {
|
|
cancel()
|
|
return err
|
|
}
|
|
|
|
snap.WatchedGateways[uid][key] = cancel
|
|
}
|
|
|
|
for key, cancelFn := range snap.WatchedGateways[uid] {
|
|
if _, ok := needGateways[key]; ok {
|
|
continue
|
|
}
|
|
gwKey := gatewayKeyFromString(key)
|
|
|
|
s.logger.Trace("stopping watch of mesh gateway",
|
|
"upstream", uid,
|
|
"chain", chain.ServiceName,
|
|
"datacenter", gwKey.Datacenter,
|
|
"partition", gwKey.Partition,
|
|
)
|
|
delete(snap.WatchedGateways[uid], key)
|
|
delete(snap.WatchedGatewayEndpoints[uid], key)
|
|
cancelFn()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
type targetWatchOpts struct {
|
|
upstreamID UpstreamID
|
|
chainID string
|
|
service string
|
|
filter string
|
|
datacenter string
|
|
entMeta *structs.EnterpriseMeta
|
|
}
|
|
|
|
func (s *handlerUpstreams) watchUpstreamTarget(ctx context.Context, snap *ConfigSnapshotUpstreams, opts targetWatchOpts) error {
|
|
s.logger.Trace("initializing watch of target",
|
|
"upstream", opts.upstreamID,
|
|
"chain", opts.service,
|
|
"target", opts.chainID,
|
|
)
|
|
|
|
var finalMeta structs.EnterpriseMeta
|
|
finalMeta.Merge(opts.entMeta)
|
|
|
|
correlationID := "upstream-target:" + opts.chainID + ":" + opts.upstreamID.String()
|
|
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
err := s.health.Notify(ctx, structs.ServiceSpecificRequest{
|
|
Datacenter: opts.datacenter,
|
|
QueryOptions: structs.QueryOptions{
|
|
Token: s.token,
|
|
Filter: opts.filter,
|
|
},
|
|
ServiceName: opts.service,
|
|
Connect: true,
|
|
// Note that Identifier doesn't type-prefix for service any more as it's
|
|
// the default and makes metrics and other things much cleaner. It's
|
|
// simpler for us if we have the type to make things unambiguous.
|
|
Source: *s.source,
|
|
EnterpriseMeta: finalMeta,
|
|
}, correlationID, s.ch)
|
|
|
|
if err != nil {
|
|
cancel()
|
|
return err
|
|
}
|
|
snap.WatchedUpstreams[opts.upstreamID][opts.chainID] = cancel
|
|
|
|
return nil
|
|
}
|
|
|
|
type discoveryChainWatchOpts struct {
|
|
id UpstreamID
|
|
name string
|
|
namespace string
|
|
partition string
|
|
datacenter string
|
|
cfg reducedUpstreamConfig
|
|
meshGateway structs.MeshGatewayConfig
|
|
}
|
|
|
|
func (s *handlerUpstreams) watchDiscoveryChain(ctx context.Context, snap *ConfigSnapshot, opts discoveryChainWatchOpts) error {
|
|
if _, ok := snap.ConnectProxy.WatchedDiscoveryChains[opts.id]; ok {
|
|
return nil
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
err := s.cache.Notify(ctx, cachetype.CompiledDiscoveryChainName, &structs.DiscoveryChainRequest{
|
|
Datacenter: s.source.Datacenter,
|
|
QueryOptions: structs.QueryOptions{Token: s.token},
|
|
Name: opts.name,
|
|
EvaluateInDatacenter: opts.datacenter,
|
|
EvaluateInNamespace: opts.namespace,
|
|
EvaluateInPartition: opts.partition,
|
|
OverrideProtocol: opts.cfg.Protocol,
|
|
OverrideConnectTimeout: opts.cfg.ConnectTimeout(),
|
|
OverrideMeshGateway: opts.meshGateway,
|
|
}, "discovery-chain:"+opts.id.String(), s.ch)
|
|
if err != nil {
|
|
cancel()
|
|
return err
|
|
}
|
|
|
|
switch s.kind {
|
|
case structs.ServiceKindIngressGateway:
|
|
snap.IngressGateway.WatchedDiscoveryChains[opts.id] = cancel
|
|
case structs.ServiceKindConnectProxy:
|
|
snap.ConnectProxy.WatchedDiscoveryChains[opts.id] = cancel
|
|
default:
|
|
cancel()
|
|
return fmt.Errorf("unsupported kind %s", s.kind)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// reducedUpstreamConfig represents the basic opaque config values that are now
|
|
// managed with the discovery chain but for backwards compatibility reasons
|
|
// should still affect how the proxy is configured.
|
|
//
|
|
// The full-blown config is agent/xds.UpstreamConfig
|
|
type reducedUpstreamConfig struct {
|
|
Protocol string `mapstructure:"protocol"`
|
|
ConnectTimeoutMs int `mapstructure:"connect_timeout_ms"`
|
|
}
|
|
|
|
func (c *reducedUpstreamConfig) ConnectTimeout() time.Duration {
|
|
return time.Duration(c.ConnectTimeoutMs) * time.Millisecond
|
|
}
|
|
|
|
func parseReducedUpstreamConfig(m map[string]interface{}) (reducedUpstreamConfig, error) {
|
|
var cfg reducedUpstreamConfig
|
|
err := mapstructure.WeakDecode(m, &cfg)
|
|
return cfg, err
|
|
}
|