mirror of https://github.com/status-im/consul.git
initial version of L7 config entry compiler (#5994)
With this you should be able to fetch all of the relevant discovery chain config entries from the state store in one query and then feed them into the compiler outside of a transaction. There are a lot of TODOs scattered through here, but they're mostly around handling fun edge cases and can be deferred until more of the plumbing works completely.
This commit is contained in:
R.B. Boyer
GitHub
parent
ceef44bbc9
commit
6a52f9f9fb
|
|
@ -0,0 +1,583 @@
|
|||
package discoverychain
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/hashicorp/consul/agent/structs"
|
||||
)
|
||||
|
||||
// TODO(rb): surface any specific errors that may matter during graph vetting at write-time (like mixing protocols)
|
||||
|
||||
// Compile assembles a discovery chain in the form of a graph of nodes using
|
||||
// raw config entries and local context.
|
||||
//
|
||||
// "Node" referenced in this file refers to a node in a graph and not to the
|
||||
// Consul construct called a "Node".
|
||||
//
|
||||
// Omitting router and splitter entries for services not using an L7 protocol
|
||||
// (like HTTP) happens during initial fetching, but for sanity purposes a quick
|
||||
// reinforcement of that happens here, too.
|
||||
func Compile(
|
||||
serviceName string,
|
||||
currentNamespace string,
|
||||
currentDatacenter string,
|
||||
inferDefaults bool,
|
||||
entries *structs.DiscoveryChainConfigEntries,
|
||||
) (*structs.CompiledDiscoveryChain, error) {
|
||||
if serviceName == "" {
|
||||
return nil, fmt.Errorf("serviceName is required")
|
||||
}
|
||||
if currentNamespace == "" {
|
||||
return nil, fmt.Errorf("currentNamespace is required")
|
||||
}
|
||||
if currentDatacenter == "" {
|
||||
return nil, fmt.Errorf("currentDatacenter is required")
|
||||
}
|
||||
if entries == nil {
|
||||
return nil, fmt.Errorf("entries is required")
|
||||
}
|
||||
|
||||
// This shouldn't be necessary, but do it anyway. It is the one place input
|
||||
// mutation will occur, but only if the caller forgot in the first place.
|
||||
entries.Fixup()
|
||||
|
||||
c := &compiler{
|
||||
serviceName: serviceName,
|
||||
currentNamespace: currentNamespace,
|
||||
currentDatacenter: currentDatacenter,
|
||||
inferDefaults: inferDefaults,
|
||||
entries: entries,
|
||||
|
||||
splitterNodes: make(map[string]*structs.DiscoveryGraphNode),
|
||||
groupResolverNodes: make(map[structs.DiscoveryTarget]*structs.DiscoveryGraphNode),
|
||||
|
||||
resolvers: make(map[string]*structs.ServiceResolverConfigEntry),
|
||||
retainResolvers: make(map[string]struct{}),
|
||||
targets: make(map[structs.DiscoveryTarget]struct{}),
|
||||
}
|
||||
|
||||
// Clone this resolver map to avoid mutating the input map during compilation.
|
||||
if len(entries.Resolvers) > 0 {
|
||||
for k, v := range entries.Resolvers {
|
||||
c.resolvers[k] = v
|
||||
}
|
||||
}
|
||||
|
||||
return c.compile()
|
||||
}
|
||||
|
||||
// compiler is a single-use struct for handling intermediate state necessary
|
||||
// for assembling a discovery chain from raw config entries.
|
||||
type compiler struct {
|
||||
serviceName string
|
||||
currentNamespace string
|
||||
currentDatacenter string
|
||||
inferDefaults bool
|
||||
|
||||
// config entries that are being compiled (will be mutated during compilation)
|
||||
//
|
||||
// This is an INPUT field.
|
||||
entries *structs.DiscoveryChainConfigEntries
|
||||
|
||||
// cached nodes
|
||||
splitterNodes map[string]*structs.DiscoveryGraphNode
|
||||
groupResolverNodes map[structs.DiscoveryTarget]*structs.DiscoveryGraphNode // this is also an OUTPUT field
|
||||
|
||||
// topNode is computed inside of assembleChain()
|
||||
//
|
||||
// This is an OUTPUT field.
|
||||
topNode *structs.DiscoveryGraphNode
|
||||
|
||||
// protocol is the common protocol used for all referenced services. These
|
||||
// cannot be mixed.
|
||||
//
|
||||
// This is an OUTPUT field.
|
||||
protocol string
|
||||
|
||||
// resolvers is initially seeded by copying the provided entries.Resolvers
|
||||
// map and default resolvers are added as they are needed.
|
||||
//
|
||||
// If redirects cause a resolver to not be needed it will be omitted from
|
||||
// this map.
|
||||
//
|
||||
// This is an OUTPUT field.
|
||||
resolvers map[string]*structs.ServiceResolverConfigEntry
|
||||
// retainResolvers flags the elements of the resolvers map that should be
|
||||
// retained in the final results.
|
||||
retainResolvers map[string]struct{}
|
||||
|
||||
// This is an OUTPUT field.
|
||||
targets map[structs.DiscoveryTarget]struct{}
|
||||
}
|
||||
|
||||
func (c *compiler) recordServiceProtocol(serviceName string) error {
|
||||
if serviceDefault, ok := c.entries.Services[serviceName]; ok {
|
||||
return c.recordProtocol(serviceName, serviceDefault.Protocol)
|
||||
} else {
|
||||
return c.recordProtocol(serviceName, "")
|
||||
}
|
||||
}
|
||||
|
||||
func (c *compiler) recordProtocol(fromService, protocol string) error {
|
||||
if protocol == "" {
|
||||
protocol = "tcp"
|
||||
} else {
|
||||
protocol = strings.ToLower(protocol)
|
||||
}
|
||||
|
||||
if c.protocol == "" {
|
||||
c.protocol = protocol
|
||||
} else if c.protocol != protocol {
|
||||
// TODO(rb): avoid this during config entry writes instead
|
||||
return fmt.Errorf("discovery chain %q uses inconsistent protocols; service %q has %q != %q", c.serviceName, fromService, protocol, c.protocol)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *compiler) compile() (*structs.CompiledDiscoveryChain, error) {
|
||||
if err := c.assembleChain(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if c.topNode == nil {
|
||||
if c.inferDefaults {
|
||||
panic("impossible to return no results with infer defaults set to true")
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
if err := c.detectCircularSplits(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := c.detectCircularResolves(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
c.flattenAdjacentSplitterNodes()
|
||||
|
||||
// Remove any unused resolvers.
|
||||
for name, _ := range c.resolvers {
|
||||
if _, ok := c.retainResolvers[name]; !ok {
|
||||
delete(c.resolvers, name)
|
||||
}
|
||||
}
|
||||
|
||||
targets := make([]structs.DiscoveryTarget, 0, len(c.targets))
|
||||
for target, _ := range c.targets {
|
||||
targets = append(targets, target)
|
||||
}
|
||||
structs.DiscoveryTargets(targets).Sort()
|
||||
|
||||
return &structs.CompiledDiscoveryChain{
|
||||
ServiceName: c.serviceName,
|
||||
Namespace: c.currentNamespace,
|
||||
Datacenter: c.currentDatacenter,
|
||||
Protocol: c.protocol,
|
||||
Node: c.topNode,
|
||||
Resolvers: c.resolvers,
|
||||
Targets: targets,
|
||||
GroupResolverNodes: c.groupResolverNodes, // TODO(rb): prune unused
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *compiler) detectCircularSplits() error {
|
||||
// TODO(rb): detect when a tree of splitters backtracks
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *compiler) detectCircularResolves() error {
|
||||
// TODO(rb): detect when a series of redirects and failovers cause a circular reference
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *compiler) flattenAdjacentSplitterNodes() {
|
||||
for {
|
||||
anyChanged := false
|
||||
for _, splitterNode := range c.splitterNodes {
|
||||
fixedSplits := make([]*structs.DiscoverySplit, 0, len(splitterNode.Splits))
|
||||
changed := false
|
||||
for _, split := range splitterNode.Splits {
|
||||
if split.Node.Type != structs.DiscoveryGraphNodeTypeSplitter {
|
||||
fixedSplits = append(fixedSplits, split)
|
||||
continue
|
||||
}
|
||||
|
||||
changed = true
|
||||
|
||||
for _, innerSplit := range split.Node.Splits {
|
||||
effectiveWeight := split.Weight * innerSplit.Weight / 100
|
||||
|
||||
newDiscoverySplit := &structs.DiscoverySplit{
|
||||
Weight: structs.NormalizeServiceSplitWeight(effectiveWeight),
|
||||
Node: innerSplit.Node,
|
||||
}
|
||||
|
||||
fixedSplits = append(fixedSplits, newDiscoverySplit)
|
||||
}
|
||||
}
|
||||
|
||||
if changed {
|
||||
splitterNode.Splits = fixedSplits
|
||||
anyChanged = true
|
||||
}
|
||||
}
|
||||
|
||||
if !anyChanged {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// assembleChain will do the initial assembly of a chain of DiscoveryGraphNode
|
||||
// entries from the provided config entries. No default resolvers are injected
|
||||
// here so it is expected that if there are no discovery chain config entries
|
||||
// set up for a given service that it will produce no topNode from this.
|
||||
func (c *compiler) assembleChain() error {
|
||||
if c.topNode != nil {
|
||||
return fmt.Errorf("assembleChain should only be called once")
|
||||
}
|
||||
|
||||
// Check for short circuit path.
|
||||
if len(c.resolvers) == 0 && c.entries.IsChainEmpty() {
|
||||
if !c.inferDefaults {
|
||||
return nil // nothing explicitly configured
|
||||
}
|
||||
|
||||
// Materialize defaults and cache.
|
||||
c.resolvers[c.serviceName] = newDefaultServiceResolver(c.serviceName)
|
||||
}
|
||||
|
||||
// The only router we consult is the one for the service name at the top of
|
||||
// the chain.
|
||||
router, ok := c.entries.Routers[c.serviceName]
|
||||
if !ok {
|
||||
// If no router is configured, move on down the line to the next hop of
|
||||
// the chain.
|
||||
node, err := c.getSplitterOrGroupResolverNode(c.newTarget(c.serviceName, "", "", ""))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c.topNode = node
|
||||
return nil
|
||||
}
|
||||
|
||||
routeNode := &structs.DiscoveryGraphNode{
|
||||
Type: structs.DiscoveryGraphNodeTypeRouter,
|
||||
Name: router.Name,
|
||||
Routes: make([]*structs.DiscoveryRoute, 0, len(router.Routes)+1),
|
||||
}
|
||||
if err := c.recordServiceProtocol(router.Name); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for i, _ := range router.Routes {
|
||||
// We don't use range variables here because we'll take the address of
|
||||
// this route and store that in a DiscoveryGraphNode and the range
|
||||
// variables share memory addresses between iterations which is exactly
|
||||
// wrong for us here.
|
||||
route := router.Routes[i]
|
||||
|
||||
compiledRoute := &structs.DiscoveryRoute{Definition: &route}
|
||||
routeNode.Routes = append(routeNode.Routes, compiledRoute)
|
||||
|
||||
dest := route.Destination
|
||||
|
||||
svc := defaultIfEmpty(dest.Service, c.serviceName)
|
||||
|
||||
// Check to see if the destination is eligible for splitting.
|
||||
var (
|
||||
node *structs.DiscoveryGraphNode
|
||||
err error
|
||||
)
|
||||
if dest.ServiceSubset == "" && dest.Namespace == "" {
|
||||
node, err = c.getSplitterOrGroupResolverNode(
|
||||
c.newTarget(svc, dest.ServiceSubset, dest.Namespace, ""),
|
||||
)
|
||||
} else {
|
||||
node, err = c.getGroupResolverNode(
|
||||
c.newTarget(svc, dest.ServiceSubset, dest.Namespace, ""),
|
||||
false,
|
||||
)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
compiledRoute.DestinationNode = node
|
||||
}
|
||||
|
||||
// If we have a router, we'll add a catch-all route at the end to send
|
||||
// unmatched traffic to the next hop in the chain.
|
||||
defaultDestinationNode, err := c.getSplitterOrGroupResolverNode(c.newTarget(c.serviceName, "", "", ""))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
defaultRoute := &structs.DiscoveryRoute{
|
||||
Definition: newDefaultServiceRoute(c.serviceName),
|
||||
DestinationNode: defaultDestinationNode,
|
||||
}
|
||||
routeNode.Routes = append(routeNode.Routes, defaultRoute)
|
||||
|
||||
c.topNode = routeNode
|
||||
return nil
|
||||
}
|
||||
|
||||
func newDefaultServiceRoute(serviceName string) *structs.ServiceRoute {
|
||||
return &structs.ServiceRoute{
|
||||
Match: &structs.ServiceRouteMatch{
|
||||
HTTP: &structs.ServiceRouteHTTPMatch{
|
||||
PathPrefix: "/",
|
||||
},
|
||||
},
|
||||
Destination: &structs.ServiceRouteDestination{
|
||||
Service: serviceName,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (c *compiler) newTarget(service, serviceSubset, namespace, datacenter string) structs.DiscoveryTarget {
|
||||
if service == "" {
|
||||
panic("newTarget called with empty service which makes no sense")
|
||||
}
|
||||
return structs.DiscoveryTarget{
|
||||
Service: service,
|
||||
ServiceSubset: serviceSubset,
|
||||
Namespace: defaultIfEmpty(namespace, c.currentNamespace),
|
||||
Datacenter: defaultIfEmpty(datacenter, c.currentDatacenter),
|
||||
}
|
||||
}
|
||||
|
||||
func (c *compiler) getSplitterOrGroupResolverNode(target structs.DiscoveryTarget) (*structs.DiscoveryGraphNode, error) {
|
||||
nextNode, err := c.getSplitterNode(target.Service)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else if nextNode != nil {
|
||||
return nextNode, nil
|
||||
}
|
||||
return c.getGroupResolverNode(target, false)
|
||||
}
|
||||
|
||||
func (c *compiler) getSplitterNode(name string) (*structs.DiscoveryGraphNode, error) {
|
||||
// Do we already have the node?
|
||||
if prev, ok := c.splitterNodes[name]; ok {
|
||||
return prev, nil
|
||||
}
|
||||
|
||||
// Fetch the config entry.
|
||||
splitter, ok := c.entries.Splitters[name]
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// Build node.
|
||||
splitNode := &structs.DiscoveryGraphNode{
|
||||
Type: structs.DiscoveryGraphNodeTypeSplitter,
|
||||
Name: name,
|
||||
Splits: make([]*structs.DiscoverySplit, 0, len(splitter.Splits)),
|
||||
}
|
||||
|
||||
// If we record this exists before recursing down it will short-circuit
|
||||
// sanely if there is some sort of graph loop below.
|
||||
c.splitterNodes[name] = splitNode
|
||||
|
||||
for _, split := range splitter.Splits {
|
||||
compiledSplit := &structs.DiscoverySplit{
|
||||
Weight: split.Weight,
|
||||
}
|
||||
splitNode.Splits = append(splitNode.Splits, compiledSplit)
|
||||
|
||||
svc := defaultIfEmpty(split.Service, name)
|
||||
// Check to see if the split is eligible for additional splitting.
|
||||
if svc != name && split.ServiceSubset == "" && split.Namespace == "" {
|
||||
nextNode, err := c.getSplitterNode(svc)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else if nextNode != nil {
|
||||
compiledSplit.Node = nextNode
|
||||
continue
|
||||
}
|
||||
// fall through to group-resolver
|
||||
}
|
||||
|
||||
node, err := c.getGroupResolverNode(
|
||||
c.newTarget(svc, split.ServiceSubset, split.Namespace, ""),
|
||||
false,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
compiledSplit.Node = node
|
||||
}
|
||||
|
||||
return splitNode, nil
|
||||
}
|
||||
|
||||
// getGroupResolverNode handles most of the code to handle
|
||||
// redirection/rewriting capabilities from a resolver config entry. It recurses
|
||||
// into itself to _generate_ targets used for failover out of convenience.
|
||||
func (c *compiler) getGroupResolverNode(target structs.DiscoveryTarget, recursedForFailover bool) (*structs.DiscoveryGraphNode, error) {
|
||||
RESOLVE_AGAIN:
|
||||
// Do we already have the node?
|
||||
if prev, ok := c.groupResolverNodes[target]; ok {
|
||||
return prev, nil
|
||||
}
|
||||
|
||||
if err := c.recordServiceProtocol(target.Service); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Fetch the config entry.
|
||||
resolver, ok := c.resolvers[target.Service]
|
||||
if !ok {
|
||||
// Materialize defaults and cache.
|
||||
resolver = newDefaultServiceResolver(target.Service)
|
||||
c.resolvers[target.Service] = resolver
|
||||
}
|
||||
|
||||
// Handle redirects right up front.
|
||||
if resolver.Redirect != nil {
|
||||
redirect := resolver.Redirect
|
||||
|
||||
redirectedTarget := target.CopyAndModify(
|
||||
redirect.Service,
|
||||
redirect.ServiceSubset,
|
||||
redirect.Namespace,
|
||||
redirect.Datacenter,
|
||||
)
|
||||
if redirectedTarget != target {
|
||||
target = redirectedTarget
|
||||
goto RESOLVE_AGAIN
|
||||
}
|
||||
}
|
||||
|
||||
// Handle default subset.
|
||||
if target.ServiceSubset == "" && resolver.DefaultSubset != "" {
|
||||
target.ServiceSubset = resolver.DefaultSubset
|
||||
goto RESOLVE_AGAIN
|
||||
}
|
||||
|
||||
// Since we're actually building a node with it, we can keep it.
|
||||
//
|
||||
// TODO(rb): maybe infer this from the keyspace of the groupresolvernodes slice.
|
||||
c.retainResolvers[target.Service] = struct{}{}
|
||||
|
||||
if target.Service != resolver.Name {
|
||||
//TODO(rb): remove
|
||||
panic("NOT POSSIBLE")
|
||||
}
|
||||
|
||||
connectTimeout := resolver.ConnectTimeout
|
||||
if connectTimeout < 1 {
|
||||
connectTimeout = 5 * time.Second
|
||||
}
|
||||
|
||||
// Build node.
|
||||
groupResolverNode := &structs.DiscoveryGraphNode{
|
||||
Type: structs.DiscoveryGraphNodeTypeGroupResolver,
|
||||
Name: resolver.Name,
|
||||
GroupResolver: &structs.DiscoveryGroupResolver{
|
||||
Definition: resolver,
|
||||
Default: resolver.IsDefault(),
|
||||
Target: target,
|
||||
ConnectTimeout: connectTimeout,
|
||||
},
|
||||
}
|
||||
groupResolver := groupResolverNode.GroupResolver
|
||||
|
||||
// Retain this target even if we may not retain the group resolver.
|
||||
c.targets[target] = struct{}{}
|
||||
|
||||
if recursedForFailover {
|
||||
// If we recursed here from ourselves in a failover context, just emit
|
||||
// this node without caching it or even processing failover again.
|
||||
// This is a little weird but it keeps the redirect/default-subset
|
||||
// logic in one place.
|
||||
return groupResolverNode, nil
|
||||
}
|
||||
|
||||
// If we record this exists before recursing down it will short-circuit
|
||||
// sanely if there is some sort of graph loop below.
|
||||
c.groupResolverNodes[target] = groupResolverNode
|
||||
|
||||
if len(resolver.Failover) > 0 {
|
||||
f := resolver.Failover
|
||||
|
||||
// Determine which failover section applies.
|
||||
failover, ok := f[target.ServiceSubset]
|
||||
if !ok {
|
||||
failover, ok = f["*"]
|
||||
}
|
||||
|
||||
if ok {
|
||||
// Determine which failover definitions apply.
|
||||
var failoverTargets []structs.DiscoveryTarget
|
||||
if len(failover.Datacenters) > 0 {
|
||||
for _, dc := range failover.Datacenters {
|
||||
// Rewrite the target as per the failover policy.
|
||||
failoverTarget := target.CopyAndModify(
|
||||
failover.Service,
|
||||
failover.ServiceSubset,
|
||||
failover.Namespace,
|
||||
dc,
|
||||
)
|
||||
if failoverTarget != target { // don't failover to yourself
|
||||
failoverTargets = append(failoverTargets, failoverTarget)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Rewrite the target as per the failover policy.
|
||||
failoverTarget := target.CopyAndModify(
|
||||
failover.Service,
|
||||
failover.ServiceSubset,
|
||||
failover.Namespace,
|
||||
"",
|
||||
)
|
||||
if failoverTarget != target { // don't failover to yourself
|
||||
failoverTargets = append(failoverTargets, failoverTarget)
|
||||
}
|
||||
}
|
||||
|
||||
// If we filtered everything out then no point in having a failover.
|
||||
if len(failoverTargets) > 0 {
|
||||
df := &structs.DiscoveryFailover{
|
||||
Definition: &failover,
|
||||
}
|
||||
groupResolver.Failover = df
|
||||
|
||||
// Convert the targets into targets by cheating a bit and
|
||||
// recursing into ourselves.
|
||||
for _, target := range failoverTargets {
|
||||
failoverGroupResolverNode, err := c.getGroupResolverNode(target, true)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if failoverGroupResolverNode.Type != structs.DiscoveryGraphNodeTypeGroupResolver {
|
||||
panic("TODO(rb)(remove): '" + failoverGroupResolverNode.Type + "' is not a group-resolver node")
|
||||
}
|
||||
failoverTarget := failoverGroupResolverNode.GroupResolver.Target
|
||||
df.Targets = append(df.Targets, failoverTarget)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return groupResolverNode, nil
|
||||
}
|
||||
|
||||
func newDefaultServiceResolver(serviceName string) *structs.ServiceResolverConfigEntry {
|
||||
return &structs.ServiceResolverConfigEntry{
|
||||
Kind: structs.ServiceResolver,
|
||||
Name: serviceName,
|
||||
}
|
||||
}
|
||||
|
||||
func defaultIfEmpty(val, defaultVal string) string {
|
||||
if val != "" {
|
||||
return val
|
||||
}
|
||||
return defaultVal
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
|
|
@ -0,0 +1,287 @@
|
|||
package state
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/hashicorp/consul/agent/structs"
|
||||
memdb "github.com/hashicorp/go-memdb"
|
||||
)
|
||||
|
||||
// ReadDiscoveryChainConfigEntries will query for the full discovery chain for
|
||||
// the provided service name. All relevant config entries will be recursively
|
||||
// fetched and included in the result.
|
||||
//
|
||||
// Once returned, the caller still needs to assemble these into a useful graph
|
||||
// structure.
|
||||
func (s *Store) ReadDiscoveryChainConfigEntries(
|
||||
ws memdb.WatchSet,
|
||||
serviceName string,
|
||||
) (uint64, *structs.DiscoveryChainConfigEntries, error) {
|
||||
tx := s.db.Txn(false)
|
||||
defer tx.Abort()
|
||||
return s.readDiscoveryChainConfigEntriesTxn(tx, ws, serviceName)
|
||||
}
|
||||
|
||||
func allowDiscoveryChainL7Features(entry *structs.ServiceConfigEntry) bool {
|
||||
if entry == nil {
|
||||
return false // default is tcp
|
||||
}
|
||||
|
||||
return structs.EnableAdvancedRoutingForProtocol(entry.Protocol)
|
||||
}
|
||||
|
||||
func (s *Store) readDiscoveryChainConfigEntriesTxn(
|
||||
tx *memdb.Txn,
|
||||
ws memdb.WatchSet,
|
||||
serviceName string,
|
||||
) (uint64, *structs.DiscoveryChainConfigEntries, error) {
|
||||
// TODO(rb): improve this so you can simulate changes to vet writes.
|
||||
|
||||
res := &structs.DiscoveryChainConfigEntries{
|
||||
Routers: make(map[string]*structs.ServiceRouterConfigEntry),
|
||||
Splitters: make(map[string]*structs.ServiceSplitterConfigEntry),
|
||||
Resolvers: make(map[string]*structs.ServiceResolverConfigEntry),
|
||||
Services: make(map[string]*structs.ServiceConfigEntry),
|
||||
}
|
||||
|
||||
// Note that below we always look up splitters and resolvers in pairs, even
|
||||
// in some circumstances where both are not strictly necessary.
|
||||
//
|
||||
// For now we'll just eat the cost of fetching pairs of splitter/resolver
|
||||
// config entries even though we may not always need both. In the common
|
||||
// case we will need the pair so there's not a big drive to optimize this
|
||||
// here at this time.
|
||||
|
||||
// Both Splitters and Resolvers maps will contain placeholder nils until
|
||||
// the end of this function to indicate "no such entry".
|
||||
|
||||
var (
|
||||
idx uint64
|
||||
activateL7 = make(map[string]struct{})
|
||||
todoSplitters = make(map[string]struct{})
|
||||
todoResolvers = make(map[string]struct{})
|
||||
)
|
||||
|
||||
checkL7 := func(name string) (bool, error) {
|
||||
if _, loaded := res.Services[name]; loaded {
|
||||
_, ok := activateL7[name]
|
||||
return ok, nil
|
||||
}
|
||||
|
||||
// first see if this is even a chain-aware protocol
|
||||
thisIdx, entry, err := s.getServiceConfigEntryTxn(tx, ws, name)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
if idx == 0 {
|
||||
idx = thisIdx
|
||||
}
|
||||
|
||||
res.Services[name] = entry // we'll strip the nil later
|
||||
if allowDiscoveryChainL7Features(entry) {
|
||||
activateL7[name] = struct{}{}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// first see if this is even a chain-aware protocol
|
||||
if useL7, err := checkL7(serviceName); err != nil {
|
||||
return 0, nil, err
|
||||
|
||||
} else if useL7 {
|
||||
// first fetch the router, of which we only collect 1 per chain eval
|
||||
_, router, err := s.getRouterConfigEntryTxn(tx, ws, serviceName)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
} else if router != nil {
|
||||
res.Routers[serviceName] = router
|
||||
}
|
||||
|
||||
if router != nil {
|
||||
for _, svc := range router.ListRelatedServices() {
|
||||
todoSplitters[svc] = struct{}{}
|
||||
}
|
||||
} else {
|
||||
// Next hop in the chain is the splitter.
|
||||
todoSplitters[serviceName] = struct{}{}
|
||||
}
|
||||
|
||||
} else {
|
||||
// Next hop in the chain is the resolver.
|
||||
res.Splitters[serviceName] = nil
|
||||
todoResolvers[serviceName] = struct{}{}
|
||||
}
|
||||
|
||||
for {
|
||||
name, ok := anyKey(todoSplitters)
|
||||
if !ok {
|
||||
break
|
||||
}
|
||||
delete(todoSplitters, name)
|
||||
|
||||
if _, ok := res.Splitters[name]; ok {
|
||||
continue // already fetched
|
||||
}
|
||||
|
||||
var splitter *structs.ServiceSplitterConfigEntry
|
||||
if useL7, err := checkL7(name); err != nil {
|
||||
return 0, nil, err
|
||||
|
||||
} else if useL7 {
|
||||
_, splitter, err = s.getSplitterConfigEntryTxn(tx, ws, name)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
|
||||
} else {
|
||||
splitter = nil // sorry
|
||||
}
|
||||
|
||||
if splitter == nil {
|
||||
res.Splitters[name] = nil
|
||||
|
||||
// Next hop in the chain is the resolver.
|
||||
todoResolvers[name] = struct{}{}
|
||||
continue
|
||||
}
|
||||
|
||||
if len(splitter.Splits) == 0 {
|
||||
return 0, nil, fmt.Errorf("found splitter config for %q that has no splits", name)
|
||||
}
|
||||
|
||||
res.Splitters[name] = splitter
|
||||
|
||||
todoResolvers[name] = struct{}{}
|
||||
for _, svc := range splitter.ListRelatedServices() {
|
||||
// If there is no splitter, this will end up adding a resolver
|
||||
// after another iteration.
|
||||
todoSplitters[svc] = struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
for {
|
||||
name, ok := anyKey(todoResolvers)
|
||||
if !ok {
|
||||
break
|
||||
}
|
||||
delete(todoResolvers, name)
|
||||
|
||||
if _, ok := res.Resolvers[name]; ok {
|
||||
continue // already fetched
|
||||
}
|
||||
|
||||
_, resolver, err := s.getResolverConfigEntryTxn(tx, ws, name)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
|
||||
if resolver == nil {
|
||||
res.Resolvers[name] = nil
|
||||
continue
|
||||
}
|
||||
|
||||
if len(resolver.Failover) > 0 {
|
||||
for subset, failoverClause := range resolver.Failover {
|
||||
if failoverClause.Service == "" &&
|
||||
failoverClause.ServiceSubset == "" &&
|
||||
failoverClause.Namespace == "" &&
|
||||
len(failoverClause.Datacenters) == 0 {
|
||||
return 0, nil, fmt.Errorf("failover section for subset %q is errantly empty", subset)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
res.Resolvers[name] = resolver
|
||||
|
||||
for _, svc := range resolver.ListRelatedServices() {
|
||||
todoResolvers[svc] = struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
res.Fixup()
|
||||
|
||||
return idx, res, nil
|
||||
}
|
||||
|
||||
// anyKey returns any key from the provided map if any exist. Useful for using
|
||||
// a map as a simple work queue of sorts.
|
||||
func anyKey(m map[string]struct{}) (string, bool) {
|
||||
if len(m) == 0 {
|
||||
return "", false
|
||||
}
|
||||
for k, _ := range m {
|
||||
return k, true
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
// getServiceConfigEntryTxn is a convenience method for fetching a
|
||||
// service-defaults kind of config entry.
|
||||
func (s *Store) getServiceConfigEntryTxn(tx *memdb.Txn, ws memdb.WatchSet, serviceName string) (uint64, *structs.ServiceConfigEntry, error) {
|
||||
idx, entry, err := s.configEntryTxn(tx, ws, structs.ServiceDefaults, serviceName)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
} else if entry == nil {
|
||||
return idx, nil, nil
|
||||
}
|
||||
|
||||
service, ok := entry.(*structs.ServiceConfigEntry)
|
||||
if !ok {
|
||||
return 0, nil, fmt.Errorf("invalid service config type %T", entry)
|
||||
}
|
||||
return idx, service, nil
|
||||
}
|
||||
|
||||
// getRouterConfigEntryTxn is a convenience method for fetching a
|
||||
// service-router kind of config entry.
|
||||
func (s *Store) getRouterConfigEntryTxn(tx *memdb.Txn, ws memdb.WatchSet, serviceName string) (uint64, *structs.ServiceRouterConfigEntry, error) {
|
||||
idx, entry, err := s.configEntryTxn(tx, ws, structs.ServiceRouter, serviceName)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
} else if entry == nil {
|
||||
return idx, nil, nil
|
||||
}
|
||||
|
||||
router, ok := entry.(*structs.ServiceRouterConfigEntry)
|
||||
if !ok {
|
||||
return 0, nil, fmt.Errorf("invalid service config type %T", entry)
|
||||
}
|
||||
return idx, router, nil
|
||||
}
|
||||
|
||||
// getSplitterConfigEntryTxn is a convenience method for fetching a
|
||||
// service-splitter kind of config entry.
|
||||
func (s *Store) getSplitterConfigEntryTxn(tx *memdb.Txn, ws memdb.WatchSet, serviceName string) (uint64, *structs.ServiceSplitterConfigEntry, error) {
|
||||
idx, entry, err := s.configEntryTxn(tx, ws, structs.ServiceSplitter, serviceName)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
} else if entry == nil {
|
||||
return idx, nil, nil
|
||||
}
|
||||
|
||||
splitter, ok := entry.(*structs.ServiceSplitterConfigEntry)
|
||||
if !ok {
|
||||
return 0, nil, fmt.Errorf("invalid service config type %T", entry)
|
||||
}
|
||||
return idx, splitter, nil
|
||||
}
|
||||
|
||||
// getResolverConfigEntryTxn is a convenience method for fetching a
|
||||
// service-resolver kind of config entry.
|
||||
func (s *Store) getResolverConfigEntryTxn(tx *memdb.Txn, ws memdb.WatchSet, serviceName string) (uint64, *structs.ServiceResolverConfigEntry, error) {
|
||||
idx, entry, err := s.configEntryTxn(tx, ws, structs.ServiceResolver, serviceName)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
} else if entry == nil {
|
||||
return idx, nil, nil
|
||||
}
|
||||
|
||||
resolver, ok := entry.(*structs.ServiceResolverConfigEntry)
|
||||
if !ok {
|
||||
return 0, nil, fmt.Errorf("invalid service config type %T", entry)
|
||||
}
|
||||
return idx, resolver, nil
|
||||
}
|
||||
|
|
@ -9,6 +9,15 @@ import (
|
|||
"github.com/hashicorp/consul/acl"
|
||||
)
|
||||
|
||||
func EnableAdvancedRoutingForProtocol(protocol string) bool {
|
||||
switch protocol {
|
||||
case "http", "http2", "grpc":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// ServiceRouterConfigEntry defines L7 (e.g. http) routing rules for a named
|
||||
// service exposed in Connect.
|
||||
//
|
||||
|
|
@ -253,18 +262,19 @@ func (e *ServiceSplitterConfigEntry) Normalize() error {
|
|||
// weight is 1/10000 or .01%
|
||||
|
||||
if len(e.Splits) > 0 {
|
||||
sumScaled := 0
|
||||
for i, split := range e.Splits {
|
||||
weightScaled := scaleWeight(split.Weight)
|
||||
e.Splits[i].Weight = float32(float32(weightScaled) / 100.0)
|
||||
|
||||
sumScaled += weightScaled
|
||||
e.Splits[i].Weight = NormalizeServiceSplitWeight(split.Weight)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func NormalizeServiceSplitWeight(weight float32) float32 {
|
||||
weightScaled := scaleWeight(weight)
|
||||
return float32(float32(weightScaled) / 100.0)
|
||||
}
|
||||
|
||||
func (e *ServiceSplitterConfigEntry) Validate() error {
|
||||
if e.Name == "" {
|
||||
return fmt.Errorf("Name is required")
|
||||
|
|
@ -441,6 +451,14 @@ type ServiceResolverConfigEntry struct {
|
|||
RaftIndex
|
||||
}
|
||||
|
||||
func (e *ServiceResolverConfigEntry) IsDefault() bool {
|
||||
return e.DefaultSubset == "" &&
|
||||
len(e.Subsets) == 0 &&
|
||||
e.Redirect == nil &&
|
||||
len(e.Failover) == 0 &&
|
||||
e.ConnectTimeout == 0
|
||||
}
|
||||
|
||||
func (e *ServiceResolverConfigEntry) GetKind() string {
|
||||
return ServiceResolver
|
||||
}
|
||||
|
|
@ -699,32 +717,10 @@ type discoveryChainConfigEntry interface {
|
|||
}
|
||||
|
||||
func canReadDiscoveryChain(entry discoveryChainConfigEntry, rule acl.Authorizer) bool {
|
||||
if rule.OperatorRead() {
|
||||
return true
|
||||
}
|
||||
|
||||
name := entry.GetName()
|
||||
|
||||
if !rule.ServiceRead(name) {
|
||||
return false
|
||||
}
|
||||
|
||||
for _, svc := range entry.ListRelatedServices() {
|
||||
if svc == name {
|
||||
continue
|
||||
}
|
||||
if !rule.ServiceRead(svc) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
return rule.ServiceRead(entry.GetName())
|
||||
}
|
||||
|
||||
func canWriteDiscoveryChain(entry discoveryChainConfigEntry, rule acl.Authorizer) bool {
|
||||
if rule.OperatorWrite() {
|
||||
return true
|
||||
}
|
||||
|
||||
name := entry.GetName()
|
||||
|
||||
if !rule.ServiceWrite(name, nil) {
|
||||
|
|
@ -744,3 +740,146 @@ func canWriteDiscoveryChain(entry discoveryChainConfigEntry, rule acl.Authorizer
|
|||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// DiscoveryChainConfigEntries wraps just the raw cross-referenced config
|
||||
// entries. None of these are defaulted.
|
||||
type DiscoveryChainConfigEntries struct {
|
||||
Routers map[string]*ServiceRouterConfigEntry
|
||||
Splitters map[string]*ServiceSplitterConfigEntry
|
||||
Resolvers map[string]*ServiceResolverConfigEntry
|
||||
Services map[string]*ServiceConfigEntry
|
||||
}
|
||||
|
||||
// Fixup ensures that the collection of entries make sense together. Nil maps
|
||||
// are initialized, nil values are deleted, and advanced features are disabled
|
||||
// if protocol dictates.
|
||||
func (e *DiscoveryChainConfigEntries) Fixup() {
|
||||
if e.Services == nil {
|
||||
e.Services = make(map[string]*ServiceConfigEntry)
|
||||
}
|
||||
if e.Routers == nil {
|
||||
e.Routers = make(map[string]*ServiceRouterConfigEntry)
|
||||
}
|
||||
if e.Splitters == nil {
|
||||
e.Splitters = make(map[string]*ServiceSplitterConfigEntry)
|
||||
}
|
||||
if e.Resolvers == nil {
|
||||
e.Resolvers = make(map[string]*ServiceResolverConfigEntry)
|
||||
}
|
||||
|
||||
for name, entry := range e.Routers {
|
||||
if entry == nil {
|
||||
delete(e.Routers, name)
|
||||
} else {
|
||||
if !e.allowAdvancedRouting(name) {
|
||||
delete(e.Routers, name)
|
||||
}
|
||||
}
|
||||
}
|
||||
for name, entry := range e.Splitters {
|
||||
if entry == nil {
|
||||
delete(e.Splitters, name)
|
||||
} else {
|
||||
if !e.allowAdvancedRouting(name) {
|
||||
delete(e.Splitters, name)
|
||||
}
|
||||
}
|
||||
}
|
||||
for name, entry := range e.Resolvers {
|
||||
if entry == nil {
|
||||
delete(e.Resolvers, name)
|
||||
}
|
||||
}
|
||||
for name, entry := range e.Services {
|
||||
if entry == nil {
|
||||
delete(e.Services, name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (e *DiscoveryChainConfigEntries) allowAdvancedRouting(name string) bool {
|
||||
if e.Services == nil {
|
||||
return false
|
||||
}
|
||||
entry, ok := e.Services[name]
|
||||
if !ok || entry == nil {
|
||||
return false
|
||||
}
|
||||
return EnableAdvancedRoutingForProtocol(entry.Protocol)
|
||||
}
|
||||
|
||||
func (e *DiscoveryChainConfigEntries) GetRouter(name string) *ServiceRouterConfigEntry {
|
||||
if e.Routers != nil {
|
||||
return e.Routers[name]
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *DiscoveryChainConfigEntries) GetSplitter(name string) *ServiceSplitterConfigEntry {
|
||||
if e.Splitters != nil {
|
||||
return e.Splitters[name]
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *DiscoveryChainConfigEntries) GetResolver(name string) *ServiceResolverConfigEntry {
|
||||
if e.Resolvers != nil {
|
||||
return e.Resolvers[name]
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *DiscoveryChainConfigEntries) GetService(name string) *ServiceConfigEntry {
|
||||
if e.Services != nil {
|
||||
return e.Services[name]
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// AddRouters adds router configs. Convenience function for testing.
|
||||
func (e *DiscoveryChainConfigEntries) AddRouters(entries ...*ServiceRouterConfigEntry) {
|
||||
if e.Routers == nil {
|
||||
e.Routers = make(map[string]*ServiceRouterConfigEntry)
|
||||
}
|
||||
for _, entry := range entries {
|
||||
e.Routers[entry.Name] = entry
|
||||
}
|
||||
}
|
||||
|
||||
// AddSplitters adds splitter configs. Convenience function for testing.
|
||||
func (e *DiscoveryChainConfigEntries) AddSplitters(entries ...*ServiceSplitterConfigEntry) {
|
||||
if e.Splitters == nil {
|
||||
e.Splitters = make(map[string]*ServiceSplitterConfigEntry)
|
||||
}
|
||||
for _, entry := range entries {
|
||||
e.Splitters[entry.Name] = entry
|
||||
}
|
||||
}
|
||||
|
||||
// AddResolvers adds resolver configs. Convenience function for testing.
|
||||
func (e *DiscoveryChainConfigEntries) AddResolvers(entries ...*ServiceResolverConfigEntry) {
|
||||
if e.Resolvers == nil {
|
||||
e.Resolvers = make(map[string]*ServiceResolverConfigEntry)
|
||||
}
|
||||
for _, entry := range entries {
|
||||
e.Resolvers[entry.Name] = entry
|
||||
}
|
||||
}
|
||||
|
||||
// AddServices adds service configs. Convenience function for testing.
|
||||
func (e *DiscoveryChainConfigEntries) AddServices(entries ...*ServiceConfigEntry) {
|
||||
if e.Services == nil {
|
||||
e.Services = make(map[string]*ServiceConfigEntry)
|
||||
}
|
||||
for _, entry := range entries {
|
||||
e.Services[entry.Name] = entry
|
||||
}
|
||||
}
|
||||
|
||||
func (e *DiscoveryChainConfigEntries) IsEmpty() bool {
|
||||
return e.IsChainEmpty() && len(e.Services) == 0
|
||||
}
|
||||
|
||||
func (e *DiscoveryChainConfigEntries) IsChainEmpty() bool {
|
||||
return len(e.Routers) == 0 && len(e.Splitters) == 0 && len(e.Resolvers) == 0
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,248 @@
|
|||
package structs
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// CompiledDiscoveryChain is the result from taking a set of related config
|
||||
// entries for a single service's discovery chain and restructuring them into a
|
||||
// form that is more usable for actual service discovery.
|
||||
type CompiledDiscoveryChain struct {
|
||||
ServiceName string
|
||||
Namespace string // the namespace that the chain was compiled within
|
||||
Datacenter string // the datacenter that the chain was compiled within
|
||||
|
||||
Protocol string // overall protocol shared by everything in the chain
|
||||
|
||||
// Node is the top node in the chain.
|
||||
//
|
||||
// If this is a router or splitter then in envoy this renders as an http
|
||||
// route object.
|
||||
//
|
||||
// If this is a group resolver then in envoy this renders as a default
|
||||
// wildcard http route object.
|
||||
Node *DiscoveryGraphNode `json:",omitempty"`
|
||||
|
||||
// GroupResolverNodes respresents all unique service instance groups that
|
||||
// need to be represented. For envoy these render as Clusters.
|
||||
//
|
||||
// Omitted from JSON because DiscoveryTarget is not a encoding.TextMarshaler.
|
||||
GroupResolverNodes map[DiscoveryTarget]*DiscoveryGraphNode `json:"-"`
|
||||
|
||||
// TODO(rb): not sure if these two fields are actually necessary but I'll know when I get into xDS
|
||||
Resolvers map[string]*ServiceResolverConfigEntry `json:",omitempty"`
|
||||
Targets []DiscoveryTarget `json:",omitempty"`
|
||||
}
|
||||
|
||||
func (c *CompiledDiscoveryChain) IsDefault() bool {
|
||||
if c.Node == nil {
|
||||
return true
|
||||
}
|
||||
return c.Node.Type == DiscoveryGraphNodeTypeGroupResolver && c.Node.GroupResolver.Default
|
||||
}
|
||||
|
||||
const (
|
||||
DiscoveryGraphNodeTypeRouter = "router"
|
||||
DiscoveryGraphNodeTypeSplitter = "splitter"
|
||||
DiscoveryGraphNodeTypeGroupResolver = "group-resolver"
|
||||
)
|
||||
|
||||
// DiscoveryGraphNode is a single node of the compiled discovery chain.
|
||||
type DiscoveryGraphNode struct {
|
||||
Type string
|
||||
Name string // default chain/service name at this spot
|
||||
|
||||
// fields for Type==router
|
||||
Routes []*DiscoveryRoute `json:",omitempty"`
|
||||
|
||||
// fields for Type==splitter
|
||||
Splits []*DiscoverySplit `json:",omitempty"`
|
||||
|
||||
// fields for Type==group-resolver
|
||||
GroupResolver *DiscoveryGroupResolver `json:",omitempty"`
|
||||
}
|
||||
|
||||
// compiled form of ServiceResolverConfigEntry but customized per non-failover target
|
||||
type DiscoveryGroupResolver struct {
|
||||
Definition *ServiceResolverConfigEntry `json:",omitempty"`
|
||||
Default bool `json:",omitempty"`
|
||||
ConnectTimeout time.Duration `json:",omitempty"`
|
||||
Target DiscoveryTarget `json:",omitempty"`
|
||||
Failover *DiscoveryFailover `json:",omitempty"`
|
||||
}
|
||||
|
||||
// compiled form of ServiceRoute
|
||||
type DiscoveryRoute struct {
|
||||
Definition *ServiceRoute `json:",omitempty"`
|
||||
DestinationNode *DiscoveryGraphNode `json:",omitempty"`
|
||||
}
|
||||
|
||||
// compiled form of ServiceSplit
|
||||
type DiscoverySplit struct {
|
||||
Weight float32 `json:",omitempty"`
|
||||
Node *DiscoveryGraphNode `json:",omitempty"`
|
||||
}
|
||||
|
||||
// compiled form of ServiceResolverFailover
|
||||
type DiscoveryFailover struct {
|
||||
Definition *ServiceResolverFailover `json:",omitempty"`
|
||||
Targets []DiscoveryTarget `json:",omitempty"`
|
||||
}
|
||||
|
||||
// DiscoveryTarget represents all of the inputs necessary to use a resolver
|
||||
// config entry to execute a catalog query to generate a list of service
|
||||
// instances during discovery.
|
||||
//
|
||||
// This is a value type so it can be used as a map key.
|
||||
type DiscoveryTarget struct {
|
||||
Service string `json:",omitempty"`
|
||||
ServiceSubset string `json:",omitempty"`
|
||||
Namespace string `json:",omitempty"`
|
||||
Datacenter string `json:",omitempty"`
|
||||
}
|
||||
|
||||
func (t DiscoveryTarget) IsEmpty() bool {
|
||||
return t.Service == "" && t.ServiceSubset == "" && t.Namespace == "" && t.Datacenter == ""
|
||||
}
|
||||
|
||||
// CopyAndModify will duplicate the target and selectively modify it given the
|
||||
// requested inputs.
|
||||
func (t DiscoveryTarget) CopyAndModify(
|
||||
service,
|
||||
serviceSubset,
|
||||
namespace,
|
||||
datacenter string,
|
||||
) DiscoveryTarget {
|
||||
t2 := t // copy
|
||||
if service != "" && service != t2.Service {
|
||||
t2.Service = service
|
||||
// Reset the chosen subset if we reference a service other than our own.
|
||||
t2.ServiceSubset = ""
|
||||
}
|
||||
if serviceSubset != "" && serviceSubset != t2.ServiceSubset {
|
||||
t2.ServiceSubset = serviceSubset
|
||||
}
|
||||
if namespace != "" && namespace != t2.Namespace {
|
||||
t2.Namespace = namespace
|
||||
}
|
||||
if datacenter != "" && datacenter != t2.Datacenter {
|
||||
t2.Datacenter = datacenter
|
||||
}
|
||||
return t2
|
||||
}
|
||||
|
||||
var _ encoding.TextMarshaler = DiscoveryTarget{}
|
||||
var _ encoding.TextUnmarshaler = (*DiscoveryTarget)(nil)
|
||||
|
||||
// MarshalText implements encoding.TextMarshaler.
|
||||
//
|
||||
// This should also not include any colons for embedding that happens
|
||||
// elsewhere.
|
||||
//
|
||||
// This should NOT return any errors.
|
||||
func (t DiscoveryTarget) MarshalText() (text []byte, err error) {
|
||||
var buf bytes.Buffer
|
||||
buf.WriteString(url.QueryEscape(t.Service))
|
||||
buf.WriteRune(',')
|
||||
buf.WriteString(url.QueryEscape(t.ServiceSubset))
|
||||
buf.WriteRune(',')
|
||||
if t.Namespace != "default" {
|
||||
buf.WriteString(url.QueryEscape(t.Namespace))
|
||||
}
|
||||
buf.WriteRune(',')
|
||||
buf.WriteString(url.QueryEscape(t.Datacenter))
|
||||
return buf.Bytes(), nil
|
||||
}
|
||||
|
||||
// UnmarshalText implements encoding.TextUnmarshaler.
|
||||
func (t *DiscoveryTarget) UnmarshalText(text []byte) error {
|
||||
parts := bytes.Split(text, []byte(","))
|
||||
bad := false
|
||||
if len(parts) != 4 {
|
||||
return fmt.Errorf("invalid target: %q", string(text))
|
||||
}
|
||||
|
||||
var err error
|
||||
t.Service, err = url.QueryUnescape(string(parts[0]))
|
||||
if err != nil {
|
||||
bad = true
|
||||
}
|
||||
t.ServiceSubset, err = url.QueryUnescape(string(parts[1]))
|
||||
if err != nil {
|
||||
bad = true
|
||||
}
|
||||
t.Namespace, err = url.QueryUnescape(string(parts[2]))
|
||||
if err != nil {
|
||||
bad = true
|
||||
}
|
||||
t.Datacenter, err = url.QueryUnescape(string(parts[3]))
|
||||
if err != nil {
|
||||
bad = true
|
||||
}
|
||||
|
||||
if bad {
|
||||
return fmt.Errorf("invalid target: %q", string(text))
|
||||
}
|
||||
|
||||
if t.Namespace == "" {
|
||||
t.Namespace = "default"
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t DiscoveryTarget) String() string {
|
||||
var b strings.Builder
|
||||
|
||||
if t.ServiceSubset != "" {
|
||||
b.WriteString(t.ServiceSubset)
|
||||
} else {
|
||||
b.WriteString("<default>")
|
||||
}
|
||||
b.WriteRune('.')
|
||||
|
||||
b.WriteString(t.Service)
|
||||
b.WriteRune('.')
|
||||
|
||||
if t.Namespace != "" {
|
||||
b.WriteString(t.Namespace)
|
||||
} else {
|
||||
b.WriteString("default")
|
||||
}
|
||||
b.WriteRune('.')
|
||||
|
||||
b.WriteString(t.Datacenter)
|
||||
|
||||
return b.String()
|
||||
}
|
||||
|
||||
type DiscoveryTargets []DiscoveryTarget
|
||||
|
||||
func (targets DiscoveryTargets) Sort() {
|
||||
sort.Slice(targets, func(i, j int) bool {
|
||||
if targets[i].Service < targets[j].Service {
|
||||
return true
|
||||
} else if targets[i].Service > targets[j].Service {
|
||||
return false
|
||||
}
|
||||
|
||||
if targets[i].ServiceSubset < targets[j].ServiceSubset {
|
||||
return true
|
||||
} else if targets[i].ServiceSubset > targets[j].ServiceSubset {
|
||||
return false
|
||||
}
|
||||
|
||||
if targets[i].Namespace < targets[j].Namespace {
|
||||
return true
|
||||
} else if targets[i].Namespace > targets[j].Namespace {
|
||||
return false
|
||||
}
|
||||
|
||||
return targets[i].Datacenter < targets[j].Datacenter
|
||||
})
|
||||
}
|
||||
|
|
@ -0,0 +1,120 @@
|
|||
package structs
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestDiscoveryTarget_TextMarshal(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
target DiscoveryTarget
|
||||
enc string
|
||||
alt DiscoveryTarget
|
||||
}{
|
||||
{
|
||||
target: DiscoveryTarget{"", "", "", ""},
|
||||
enc: ",,,",
|
||||
alt: DiscoveryTarget{"", "", "default", ""},
|
||||
},
|
||||
{
|
||||
target: DiscoveryTarget{"a:b", "", "", ""},
|
||||
enc: "a%3Ab,,,",
|
||||
alt: DiscoveryTarget{"a:b", "", "default", ""},
|
||||
},
|
||||
{
|
||||
target: DiscoveryTarget{"", "a:b", "", ""},
|
||||
enc: ",a%3Ab,,",
|
||||
alt: DiscoveryTarget{"", "a:b", "default", ""},
|
||||
},
|
||||
{
|
||||
target: DiscoveryTarget{"", "", "a:b", ""},
|
||||
enc: ",,a%3Ab,",
|
||||
alt: DiscoveryTarget{"", "", "a:b", ""},
|
||||
},
|
||||
{
|
||||
target: DiscoveryTarget{"", "", "", "a:b"},
|
||||
enc: ",,,a%3Ab",
|
||||
alt: DiscoveryTarget{"", "", "default", "a:b"},
|
||||
},
|
||||
{
|
||||
target: DiscoveryTarget{"one", "two", "three", "four"},
|
||||
enc: "one,two,three,four",
|
||||
},
|
||||
} {
|
||||
tc := tc
|
||||
t.Run(tc.target.String(), func(t *testing.T) {
|
||||
out, err := tc.target.MarshalText()
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, tc.enc, string(out))
|
||||
|
||||
var dec DiscoveryTarget
|
||||
require.NoError(t, dec.UnmarshalText(out))
|
||||
if tc.alt.IsEmpty() {
|
||||
require.Equal(t, tc.target, dec)
|
||||
} else {
|
||||
require.Equal(t, tc.alt, dec)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDiscoveryTarget_CopyAndModify(t *testing.T) {
|
||||
type fields = DiscoveryTarget // abbreviation
|
||||
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
in fields
|
||||
mod fields // this is semantically wrong, but the shape of the struct is still what we want
|
||||
expect fields
|
||||
}{
|
||||
{
|
||||
name: "service with no subset and no mod",
|
||||
in: fields{"foo", "", "default", "dc1"},
|
||||
mod: fields{},
|
||||
expect: fields{"foo", "", "default", "dc1"},
|
||||
},
|
||||
{
|
||||
name: "service with subset and no mod",
|
||||
in: fields{"foo", "v2", "default", "dc1"},
|
||||
mod: fields{},
|
||||
expect: fields{"foo", "v2", "default", "dc1"},
|
||||
},
|
||||
{
|
||||
name: "service with no subset and service mod",
|
||||
in: fields{"foo", "", "default", "dc1"},
|
||||
mod: fields{"bar", "", "", ""},
|
||||
expect: fields{"bar", "", "default", "dc1"},
|
||||
},
|
||||
{
|
||||
name: "service with subset and service mod",
|
||||
in: fields{"foo", "v2", "default", "dc1"},
|
||||
mod: fields{"bar", "", "", ""},
|
||||
expect: fields{"bar", "", "default", "dc1"},
|
||||
},
|
||||
{
|
||||
name: "service with subset and noop service mod with dc mod",
|
||||
in: fields{"foo", "v2", "default", "dc1"},
|
||||
mod: fields{"foo", "", "", "dc9"},
|
||||
expect: fields{"foo", "v2", "default", "dc9"},
|
||||
},
|
||||
{
|
||||
name: "service with subset and namespace mod",
|
||||
in: fields{"foo", "v2", "default", "dc1"},
|
||||
mod: fields{"", "", "fancy", ""},
|
||||
expect: fields{"foo", "v2", "fancy", "dc1"},
|
||||
},
|
||||
} {
|
||||
tc := tc
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
out := tc.in.CopyAndModify(
|
||||
tc.mod.Service,
|
||||
tc.mod.ServiceSubset,
|
||||
tc.mod.Namespace,
|
||||
tc.mod.Datacenter,
|
||||
)
|
||||
require.Equal(t, tc.expect, out)
|
||||
})
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue