open-consul/agent/consul/catalog_endpoint.go

370 lines
10 KiB
Go
Raw Normal View History

package consul
import (
2013-12-24 20:43:34 +00:00
"fmt"
2014-02-20 23:16:26 +00:00
"time"
"github.com/armon/go-metrics"
"github.com/hashicorp/consul/acl"
pkg refactor command/agent/* -> agent/* command/consul/* -> agent/consul/* command/agent/command{,_test}.go -> command/agent{,_test}.go command/base/command.go -> command/base.go command/base/* -> command/* commands.go -> command/commands.go The script which did the refactor is: ( cd $GOPATH/src/github.com/hashicorp/consul git mv command/agent/command.go command/agent.go git mv command/agent/command_test.go command/agent_test.go git mv command/agent/flag_slice_value{,_test}.go command/ git mv command/agent . git mv command/base/command.go command/base.go git mv command/base/config_util{,_test}.go command/ git mv commands.go command/ git mv consul agent rmdir command/base/ gsed -i -e 's|package agent|package command|' command/agent{,_test}.go gsed -i -e 's|package agent|package command|' command/flag_slice_value{,_test}.go gsed -i -e 's|package base|package command|' command/base.go command/config_util{,_test}.go gsed -i -e 's|package main|package command|' command/commands.go gsed -i -e 's|base.Command|BaseCommand|' command/commands.go gsed -i -e 's|agent.Command|AgentCommand|' command/commands.go gsed -i -e 's|\tCommand:|\tBaseCommand:|' command/commands.go gsed -i -e 's|base\.||' command/commands.go gsed -i -e 's|command\.||' command/commands.go gsed -i -e 's|command|c|' main.go gsed -i -e 's|range Commands|range command.Commands|' main.go gsed -i -e 's|Commands: Commands|Commands: command.Commands|' main.go gsed -i -e 's|base\.BoolValue|BoolValue|' command/operator_autopilot_set.go gsed -i -e 's|base\.DurationValue|DurationValue|' command/operator_autopilot_set.go gsed -i -e 's|base\.StringValue|StringValue|' command/operator_autopilot_set.go gsed -i -e 's|base\.UintValue|UintValue|' command/operator_autopilot_set.go gsed -i -e 's|\bCommand\b|BaseCommand|' command/base.go gsed -i -e 's|BaseCommand Options|Command Options|' command/base.go gsed -i -e 's|base.Command|BaseCommand|' command/*.go gsed -i -e 's|c\.Command|c.BaseCommand|g' command/*.go gsed -i -e 's|\tCommand:|\tBaseCommand:|' command/*_test.go gsed -i -e 's|base\.||' command/*_test.go gsed -i -e 's|\bCommand\b|AgentCommand|' command/agent{,_test}.go gsed -i -e 's|cmd.AgentCommand|cmd.BaseCommand|' command/agent.go gsed -i -e 's|cli.AgentCommand = new(Command)|cli.Command = new(AgentCommand)|' command/agent_test.go gsed -i -e 's|exec.AgentCommand|exec.Command|' command/agent_test.go gsed -i -e 's|exec.BaseCommand|exec.Command|' command/agent_test.go gsed -i -e 's|NewTestAgent|agent.NewTestAgent|' command/agent_test.go gsed -i -e 's|= TestConfig|= agent.TestConfig|' command/agent_test.go gsed -i -e 's|: RetryJoin|: agent.RetryJoin|' command/agent_test.go gsed -i -e 's|\.\./\.\./|../|' command/config_util_test.go gsed -i -e 's|\bverifyUniqueListeners|VerifyUniqueListeners|' agent/config{,_test}.go command/agent.go gsed -i -e 's|\bserfLANKeyring\b|SerfLANKeyring|g' agent/{agent,keyring,testagent}.go command/agent.go gsed -i -e 's|\bserfWANKeyring\b|SerfWANKeyring|g' agent/{agent,keyring,testagent}.go command/agent.go gsed -i -e 's|\bNewAgent\b|agent.New|g' command/agent{,_test}.go gsed -i -e 's|\bNewAgent|New|' agent/{acl_test,agent,testagent}.go gsed -i -e 's|\bAgent\b|agent.&|g' command/agent{,_test}.go gsed -i -e 's|\bBool\b|agent.&|g' command/agent{,_test}.go gsed -i -e 's|\bConfig\b|agent.&|g' command/agent{,_test}.go gsed -i -e 's|\bDefaultConfig\b|agent.&|g' command/agent{,_test}.go gsed -i -e 's|\bDevConfig\b|agent.&|g' command/agent{,_test}.go gsed -i -e 's|\bMergeConfig\b|agent.&|g' command/agent{,_test}.go gsed -i -e 's|\bReadConfigPaths\b|agent.&|g' command/agent{,_test}.go gsed -i -e 's|\bParseMetaPair\b|agent.&|g' command/agent{,_test}.go gsed -i -e 's|\bSerfLANKeyring\b|agent.&|g' command/agent{,_test}.go gsed -i -e 's|\bSerfWANKeyring\b|agent.&|g' command/agent{,_test}.go gsed -i -e 's|circonus\.agent|circonus|g' command/agent{,_test}.go gsed -i -e 's|logger\.agent|logger|g' command/agent{,_test}.go gsed -i -e 's|metrics\.agent|metrics|g' command/agent{,_test}.go gsed -i -e 's|// agent.Agent|// agent|' command/agent{,_test}.go gsed -i -e 's|a\.agent\.Config|a.Config|' command/agent{,_test}.go gsed -i -e 's|agent\.AppendSliceValue|AppendSliceValue|' command/{configtest,validate}.go gsed -i -e 's|consul/consul|agent/consul|' GNUmakefile gsed -i -e 's|\.\./test|../../test|' agent/consul/server_test.go # fix imports f=$(grep -rl 'github.com/hashicorp/consul/command/agent' * | grep '\.go') gsed -i -e 's|github.com/hashicorp/consul/command/agent|github.com/hashicorp/consul/agent|' $f goimports -w $f f=$(grep -rl 'github.com/hashicorp/consul/consul' * | grep '\.go') gsed -i -e 's|github.com/hashicorp/consul/consul|github.com/hashicorp/consul/agent/consul|' $f goimports -w $f goimports -w command/*.go main.go )
2017-06-09 22:28:28 +00:00
"github.com/hashicorp/consul/agent/consul/state"
"github.com/hashicorp/consul/agent/structs"
"github.com/hashicorp/consul/ipaddr"
"github.com/hashicorp/consul/types"
"github.com/hashicorp/go-memdb"
2017-01-18 22:26:42 +00:00
"github.com/hashicorp/go-uuid"
)
// Catalog endpoint is used to manipulate the service catalog
type Catalog struct {
2013-12-11 22:57:40 +00:00
srv *Server
}
// Register is used register that a node is providing a given service.
2013-12-19 20:03:57 +00:00
func (c *Catalog) Register(args *structs.RegisterRequest, reply *struct{}) error {
2014-04-19 00:17:12 +00:00
if done, err := c.srv.forward("Catalog.Register", args, args, reply); done {
return err
}
defer metrics.MeasureSince([]string{"catalog", "register"}, time.Now())
// Verify the args.
2017-11-01 21:25:46 +00:00
if args.Node == "" {
return fmt.Errorf("Must provide node")
}
if args.Address == "" && !args.SkipNodeUpdate {
return fmt.Errorf("Must provide address if SkipNodeUpdate is not set")
2013-12-24 20:43:34 +00:00
}
2017-01-18 22:26:42 +00:00
if args.ID != "" {
if _, err := uuid.ParseUUID(string(args.ID)); err != nil {
return fmt.Errorf("Bad node ID: %v", err)
}
}
2013-12-24 20:43:34 +00:00
// Fetch the ACL token, if any.
rule, err := c.srv.resolveToken(args.Token)
if err != nil {
return err
}
// Handle a service registration.
if args.Service != nil {
// Validate the service. This is in addition to the below since
// the above just hasn't been moved over yet. We should move it over
// in time.
if err := args.Service.Validate(); err != nil {
return err
}
// If no service id, but service name, use default
if args.Service.ID == "" && args.Service.Service != "" {
args.Service.ID = args.Service.Service
}
// Verify ServiceName provided if ID.
if args.Service.ID != "" && args.Service.Service == "" {
return fmt.Errorf("Must provide service name with ID")
}
// Check the service address here and in the agent endpoint
// since service registration isn't synchronous.
if ipaddr.IsAny(args.Service.Address) {
return fmt.Errorf("Invalid service address")
}
// Apply the ACL policy if any. The 'consul' service is excluded
// since it is managed automatically internally (that behavior
// is going away after version 0.8). We check this same policy
// later if version 0.8 is enabled, so we can eventually just
// delete this and do all the ACL checks down there.
if args.Service.Service != structs.ConsulServiceName {
if rule != nil && !rule.ServiceWrite(args.Service.Service, nil) {
return acl.ErrPermissionDenied
}
}
// Proxies must have write permission on their destination
if args.Service.Kind == structs.ServiceKindConnectProxy {
if rule != nil && !rule.ServiceWrite(args.Service.ProxyDestination, nil) {
return acl.ErrPermissionDenied
}
}
}
// Move the old format single check into the slice, and fixup IDs.
if args.Check != nil {
args.Checks = append(args.Checks, args.Check)
args.Check = nil
}
for _, check := range args.Checks {
if check.CheckID == "" && check.Name != "" {
check.CheckID = types.CheckID(check.Name)
}
if check.Node == "" {
check.Node = args.Node
}
}
// Check the complete register request against the given ACL policy.
if rule != nil && c.srv.config.ACLEnforceVersion8 {
state := c.srv.fsm.State()
_, ns, err := state.NodeServices(nil, args.Node)
if err != nil {
return fmt.Errorf("Node lookup failed: %v", err)
}
if err := vetRegisterWithACL(rule, args, ns); err != nil {
return err
}
}
resp, err := c.srv.raftApply(structs.RegisterRequestType, args)
if err != nil {
return err
}
if respErr, ok := resp.(error); ok {
return respErr
}
return nil
}
// Deregister is used to remove a service registration for a given node.
2013-12-19 20:03:57 +00:00
func (c *Catalog) Deregister(args *structs.DeregisterRequest, reply *struct{}) error {
2014-04-19 00:17:12 +00:00
if done, err := c.srv.forward("Catalog.Deregister", args, args, reply); done {
2013-12-11 23:34:10 +00:00
return err
}
defer metrics.MeasureSince([]string{"catalog", "deregister"}, time.Now())
2013-12-11 23:34:10 +00:00
2013-12-24 20:43:34 +00:00
// Verify the args
if args.Node == "" {
return fmt.Errorf("Must provide node")
}
// Fetch the ACL token, if any.
rule, err := c.srv.resolveToken(args.Token)
2013-12-11 23:34:10 +00:00
if err != nil {
return err
}
// Check the complete deregister request against the given ACL policy.
if rule != nil && c.srv.config.ACLEnforceVersion8 {
state := c.srv.fsm.State()
var ns *structs.NodeService
if args.ServiceID != "" {
_, ns, err = state.NodeService(args.Node, args.ServiceID)
if err != nil {
return fmt.Errorf("Service lookup failed: %v", err)
}
}
var nc *structs.HealthCheck
if args.CheckID != "" {
_, nc, err = state.NodeCheck(args.Node, args.CheckID)
if err != nil {
return fmt.Errorf("Check lookup failed: %v", err)
}
}
if err := vetDeregisterWithACL(rule, args, ns, nc); err != nil {
return err
}
}
if _, err := c.srv.raftApply(structs.DeregisterRequestType, args); err != nil {
2013-12-11 23:34:10 +00:00
return err
}
return nil
}
2013-12-12 18:35:50 +00:00
// ListDatacenters is used to query for the list of known datacenters
func (c *Catalog) ListDatacenters(args *struct{}, reply *[]string) error {
dcs, err := c.srv.router.GetDatacentersByDistance()
if err != nil {
return err
}
if len(dcs) == 0 { // no WAN federation, so return the local data center name
dcs = []string{c.srv.config.Datacenter}
}
2013-12-12 18:35:50 +00:00
*reply = dcs
return nil
}
2013-12-12 18:48:36 +00:00
// ListNodes is used to query the nodes in a DC
func (c *Catalog) ListNodes(args *structs.DCSpecificRequest, reply *structs.IndexedNodes) error {
2014-04-19 00:17:12 +00:00
if done, err := c.srv.forward("Catalog.ListNodes", args, args, reply); done {
2013-12-12 18:48:36 +00:00
return err
}
return c.srv.blockingQuery(
&args.QueryOptions,
&reply.QueryMeta,
2017-04-21 00:46:29 +00:00
func(ws memdb.WatchSet, state *state.Store) error {
var index uint64
var nodes structs.Nodes
var err error
if len(args.NodeMetaFilters) > 0 {
index, nodes, err = state.NodesByMeta(ws, args.NodeMetaFilters)
} else {
index, nodes, err = state.Nodes(ws)
}
if err != nil {
return err
}
reply.Index, reply.Nodes = index, nodes
if err := c.srv.filterACL(args.Token, reply); err != nil {
return err
}
return c.srv.sortNodesByDistanceFrom(args.Source, reply.Nodes)
})
2013-12-12 18:48:36 +00:00
}
2013-12-12 19:07:14 +00:00
// ListServices is used to query the services in a DC
func (c *Catalog) ListServices(args *structs.DCSpecificRequest, reply *structs.IndexedServices) error {
2014-04-19 00:17:12 +00:00
if done, err := c.srv.forward("Catalog.ListServices", args, args, reply); done {
2013-12-12 19:07:14 +00:00
return err
}
return c.srv.blockingQuery(
&args.QueryOptions,
&reply.QueryMeta,
2017-04-21 00:46:29 +00:00
func(ws memdb.WatchSet, state *state.Store) error {
var index uint64
var services structs.Services
var err error
if len(args.NodeMetaFilters) > 0 {
index, services, err = state.ServicesByNodeMeta(ws, args.NodeMetaFilters)
} else {
index, services, err = state.Services(ws)
}
if err != nil {
return err
}
reply.Index, reply.Services = index, services
return c.srv.filterACL(args.Token, reply)
})
2013-12-12 19:07:14 +00:00
}
2013-12-12 19:37:19 +00:00
// ServiceNodes returns all the nodes registered as part of a service
func (c *Catalog) ServiceNodes(args *structs.ServiceSpecificRequest, reply *structs.IndexedServiceNodes) error {
2014-04-19 00:17:12 +00:00
if done, err := c.srv.forward("Catalog.ServiceNodes", args, args, reply); done {
2013-12-12 19:37:19 +00:00
return err
}
2013-12-24 20:43:34 +00:00
// Verify the arguments
if args.ServiceName == "" && args.ServiceAddress == "" {
2013-12-24 20:43:34 +00:00
return fmt.Errorf("Must provide service name")
}
// Determine the function we'll call
var f func(memdb.WatchSet, *state.Store) (uint64, structs.ServiceNodes, error)
switch {
case args.Connect:
f = func(ws memdb.WatchSet, s *state.Store) (uint64, structs.ServiceNodes, error) {
return s.ConnectServiceNodes(ws, args.ServiceName)
}
default:
f = func(ws memdb.WatchSet, s *state.Store) (uint64, structs.ServiceNodes, error) {
if args.ServiceAddress != "" {
return s.ServiceAddressNodes(ws, args.ServiceAddress)
}
if args.TagFilter {
return s.ServiceTagNodes(ws, args.ServiceName, args.ServiceTag)
}
return s.ServiceNodes(ws, args.ServiceName)
}
}
// If we're doing a connect query, we need read access to the service
// we're trying to find proxies for, so check that.
if args.Connect {
// Fetch the ACL token, if any.
rule, err := c.srv.resolveToken(args.Token)
if err != nil {
return err
}
if rule != nil && !rule.ServiceRead(args.ServiceName) {
// Just return nil, which will return an empty response (tested)
return nil
}
}
err := c.srv.blockingQuery(
&args.QueryOptions,
&reply.QueryMeta,
2017-04-21 00:46:29 +00:00
func(ws memdb.WatchSet, state *state.Store) error {
index, services, err := f(ws, state)
if err != nil {
return err
}
reply.Index, reply.ServiceNodes = index, services
if len(args.NodeMetaFilters) > 0 {
var filtered structs.ServiceNodes
for _, service := range services {
if structs.SatisfiesMetaFilters(service.NodeMeta, args.NodeMetaFilters) {
filtered = append(filtered, service)
}
}
reply.ServiceNodes = filtered
}
if err := c.srv.filterACL(args.Token, reply); err != nil {
return err
}
return c.srv.sortNodesByDistanceFrom(args.Source, reply.ServiceNodes)
})
// Provide some metrics
if err == nil {
// For metrics, we separate Connect-based lookups from non-Connect
key := "service"
if args.Connect {
key = "connect"
}
metrics.IncrCounterWithLabels([]string{"catalog", key, "query"}, 1,
[]metrics.Label{{Name: "service", Value: args.ServiceName}})
if args.ServiceTag != "" {
metrics.IncrCounterWithLabels([]string{"catalog", key, "query-tag"}, 1,
[]metrics.Label{{Name: "service", Value: args.ServiceName}, {Name: "tag", Value: args.ServiceTag}})
}
if len(reply.ServiceNodes) == 0 {
metrics.IncrCounterWithLabels([]string{"catalog", key, "not-found"}, 1,
[]metrics.Label{{Name: "service", Value: args.ServiceName}})
}
}
return err
2013-12-12 19:37:19 +00:00
}
2013-12-12 19:46:25 +00:00
// NodeServices returns all the services registered as part of a node
func (c *Catalog) NodeServices(args *structs.NodeSpecificRequest, reply *structs.IndexedNodeServices) error {
2014-04-19 00:17:12 +00:00
if done, err := c.srv.forward("Catalog.NodeServices", args, args, reply); done {
2013-12-12 19:46:25 +00:00
return err
}
2013-12-24 20:43:34 +00:00
// Verify the arguments
if args.Node == "" {
return fmt.Errorf("Must provide node")
}
return c.srv.blockingQuery(
&args.QueryOptions,
&reply.QueryMeta,
2017-04-21 00:46:29 +00:00
func(ws memdb.WatchSet, state *state.Store) error {
index, services, err := state.NodeServices(ws, args.Node)
if err != nil {
return err
}
reply.Index, reply.NodeServices = index, services
return c.srv.filterACL(args.Token, reply)
})
2013-12-12 19:46:25 +00:00
}