mirror of
https://github.com/status-im/consul.git
synced 2025-01-12 23:05:28 +00:00
13c04a13af
This is the OSS portion of enterprise PR 2339. It improves our handling of "irrecoverable" errors in proxycfg data sources. The canonical example of this is what happens when the ACL token presented by Envoy is deleted/revoked. Previously, the stream would get "stuck" until the xDS server re-checked the token (after 5 minutes) and terminated the stream. Materializers would also sit burning resources retrying something that could never succeed. Now, it is possible for data sources to mark errors as "terminal" which causes the xDS stream to be closed immediately. Similarly, the submatview.Store will evict materializers when it observes they have encountered such an error.
145 lines
6.4 KiB
Go
145 lines
6.4 KiB
Go
package proxycfgglue
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/hashicorp/consul/proto/pbpeering"
|
|
"github.com/hashicorp/go-memdb"
|
|
|
|
"github.com/hashicorp/consul/acl"
|
|
"github.com/hashicorp/consul/agent/cache"
|
|
cachetype "github.com/hashicorp/consul/agent/cache-types"
|
|
"github.com/hashicorp/consul/agent/configentry"
|
|
"github.com/hashicorp/consul/agent/consul/discoverychain"
|
|
"github.com/hashicorp/consul/agent/consul/state"
|
|
"github.com/hashicorp/consul/agent/consul/watch"
|
|
"github.com/hashicorp/consul/agent/proxycfg"
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
)
|
|
|
|
// Store is the state store interface required for server-local data sources.
|
|
type Store interface {
|
|
watch.StateStore
|
|
|
|
ExportedServicesForAllPeersByName(ws memdb.WatchSet, entMeta acl.EnterpriseMeta) (uint64, map[string]structs.ServiceList, error)
|
|
FederationStateList(ws memdb.WatchSet) (uint64, []*structs.FederationState, error)
|
|
GatewayServices(ws memdb.WatchSet, gateway string, entMeta *acl.EnterpriseMeta) (uint64, structs.GatewayServices, error)
|
|
IntentionTopology(ws memdb.WatchSet, target structs.ServiceName, downstreams bool, defaultDecision acl.EnforcementDecision, intentionTarget structs.IntentionTargetType) (uint64, structs.ServiceList, error)
|
|
ServiceDiscoveryChain(ws memdb.WatchSet, serviceName string, entMeta *acl.EnterpriseMeta, req discoverychain.CompileRequest) (uint64, *structs.CompiledDiscoveryChain, *configentry.DiscoveryChainSet, error)
|
|
PeeringTrustBundleRead(ws memdb.WatchSet, q state.Query) (uint64, *pbpeering.PeeringTrustBundle, error)
|
|
PeeringTrustBundleList(ws memdb.WatchSet, entMeta acl.EnterpriseMeta) (uint64, []*pbpeering.PeeringTrustBundle, error)
|
|
TrustBundleListByService(ws memdb.WatchSet, service, dc string, entMeta acl.EnterpriseMeta) (uint64, []*pbpeering.PeeringTrustBundle, error)
|
|
VirtualIPsForAllImportedServices(ws memdb.WatchSet, entMeta acl.EnterpriseMeta) (uint64, []state.ServiceVirtualIP, error)
|
|
}
|
|
|
|
// CacheCARoots satisfies the proxycfg.CARoots interface by sourcing data from
|
|
// the agent cache.
|
|
func CacheCARoots(c *cache.Cache) proxycfg.CARoots {
|
|
return &cacheProxyDataSource[*structs.DCSpecificRequest]{c, cachetype.ConnectCARootName}
|
|
}
|
|
|
|
// CacheConfigEntry satisfies the proxycfg.ConfigEntry interface by sourcing
|
|
// data from the agent cache.
|
|
func CacheConfigEntry(c *cache.Cache) proxycfg.ConfigEntry {
|
|
return &cacheProxyDataSource[*structs.ConfigEntryQuery]{c, cachetype.ConfigEntryName}
|
|
}
|
|
|
|
// CacheConfigEntryList satisfies the proxycfg.ConfigEntryList interface by
|
|
// sourcing data from the agent cache.
|
|
func CacheConfigEntryList(c *cache.Cache) proxycfg.ConfigEntryList {
|
|
return &cacheProxyDataSource[*structs.ConfigEntryQuery]{c, cachetype.ConfigEntryListName}
|
|
}
|
|
|
|
// CacheDatacenters satisfies the proxycfg.Datacenters interface by sourcing
|
|
// data from the agent cache.
|
|
func CacheDatacenters(c *cache.Cache) proxycfg.Datacenters {
|
|
return &cacheProxyDataSource[*structs.DatacentersRequest]{c, cachetype.CatalogDatacentersName}
|
|
}
|
|
|
|
// CacheServiceGateways satisfies the proxycfg.ServiceGateways interface by
|
|
// sourcing data from the agent cache.
|
|
func CacheServiceGateways(c *cache.Cache) proxycfg.GatewayServices {
|
|
return &cacheProxyDataSource[*structs.ServiceSpecificRequest]{c, cachetype.ServiceGatewaysName}
|
|
}
|
|
|
|
// CacheHTTPChecks satisifies the proxycfg.HTTPChecks interface by sourcing
|
|
// data from the agent cache.
|
|
func CacheHTTPChecks(c *cache.Cache) proxycfg.HTTPChecks {
|
|
return &cacheProxyDataSource[*cachetype.ServiceHTTPChecksRequest]{c, cachetype.ServiceHTTPChecksName}
|
|
}
|
|
|
|
// CacheIntentionUpstreams satisfies the proxycfg.IntentionUpstreams interface
|
|
// by sourcing data from the agent cache.
|
|
func CacheIntentionUpstreams(c *cache.Cache) proxycfg.IntentionUpstreams {
|
|
return &cacheProxyDataSource[*structs.ServiceSpecificRequest]{c, cachetype.IntentionUpstreamsName}
|
|
}
|
|
|
|
// CacheIntentionUpstreamsDestination satisfies the proxycfg.IntentionUpstreamsDestination interface
|
|
// by sourcing data from the agent cache.
|
|
func CacheIntentionUpstreamsDestination(c *cache.Cache) proxycfg.IntentionUpstreams {
|
|
return &cacheProxyDataSource[*structs.ServiceSpecificRequest]{c, cachetype.IntentionUpstreamsDestinationName}
|
|
}
|
|
|
|
// CacheInternalServiceDump satisfies the proxycfg.InternalServiceDump
|
|
// interface by sourcing data from the agent cache.
|
|
func CacheInternalServiceDump(c *cache.Cache) proxycfg.InternalServiceDump {
|
|
return &cacheProxyDataSource[*structs.ServiceDumpRequest]{c, cachetype.InternalServiceDumpName}
|
|
}
|
|
|
|
// CacheLeafCertificate satisifies the proxycfg.LeafCertificate interface by
|
|
// sourcing data from the agent cache.
|
|
func CacheLeafCertificate(c *cache.Cache) proxycfg.LeafCertificate {
|
|
return &cacheProxyDataSource[*cachetype.ConnectCALeafRequest]{c, cachetype.ConnectCALeafName}
|
|
}
|
|
|
|
// CachePrepraredQuery satisfies the proxycfg.PreparedQuery interface by
|
|
// sourcing data from the agent cache.
|
|
func CachePrepraredQuery(c *cache.Cache) proxycfg.PreparedQuery {
|
|
return &cacheProxyDataSource[*structs.PreparedQueryExecuteRequest]{c, cachetype.PreparedQueryName}
|
|
}
|
|
|
|
// CacheResolvedServiceConfig satisfies the proxycfg.ResolvedServiceConfig
|
|
// interface by sourcing data from the agent cache.
|
|
func CacheResolvedServiceConfig(c *cache.Cache) proxycfg.ResolvedServiceConfig {
|
|
return &cacheProxyDataSource[*structs.ServiceConfigRequest]{c, cachetype.ResolvedServiceConfigName}
|
|
}
|
|
|
|
// cacheProxyDataSource implements a generic wrapper around the agent cache to
|
|
// provide data to the proxycfg.Manager.
|
|
type cacheProxyDataSource[ReqType cache.Request] struct {
|
|
c *cache.Cache
|
|
t string
|
|
}
|
|
|
|
// Notify satisfies the interfaces used by proxycfg.Manager to source data by
|
|
// subscribing to notifications from the agent cache.
|
|
func (c *cacheProxyDataSource[ReqType]) Notify(
|
|
ctx context.Context,
|
|
req ReqType,
|
|
correlationID string,
|
|
ch chan<- proxycfg.UpdateEvent,
|
|
) error {
|
|
return c.c.NotifyCallback(ctx, c.t, req, correlationID, dispatchCacheUpdate(ch))
|
|
}
|
|
|
|
func dispatchCacheUpdate(ch chan<- proxycfg.UpdateEvent) cache.Callback {
|
|
return func(ctx context.Context, e cache.UpdateEvent) {
|
|
select {
|
|
case ch <- newUpdateEvent(e.CorrelationID, e.Result, e.Err):
|
|
case <-ctx.Done():
|
|
}
|
|
}
|
|
}
|
|
|
|
func newUpdateEvent(correlationID string, result any, err error) proxycfg.UpdateEvent {
|
|
// This roughly matches the logic in agent/submatview.LocalMaterializer.isTerminalError.
|
|
if acl.IsErrNotFound(err) {
|
|
err = proxycfg.TerminalError(err)
|
|
}
|
|
return proxycfg.UpdateEvent{
|
|
CorrelationID: correlationID,
|
|
Result: result,
|
|
Err: err,
|
|
}
|
|
}
|