d63c5807cf
UUID auto-generation here causes trouble in a few cases. The biggest being older nodes reregistering will fail when the UUIDs are different and the names match This reverts commit 0f700340828f464449c2e0d5a82db0bc5456d385. This reverts commit d1a8f9cb3f6f48dd9c8d0bc858031ff6ccff51d0. This reverts commit cf69ec42a418ab6594a6654e9545e12160f30970.
370 lines
10 KiB
Go
370 lines
10 KiB
Go
package consul
|
|
|
|
import (
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/armon/go-metrics"
|
|
"github.com/hashicorp/consul/acl"
|
|
"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"
|
|
"github.com/hashicorp/go-uuid"
|
|
)
|
|
|
|
// Catalog endpoint is used to manipulate the service catalog
|
|
type Catalog struct {
|
|
srv *Server
|
|
}
|
|
|
|
// Register is used register that a node is providing a given service.
|
|
func (c *Catalog) Register(args *structs.RegisterRequest, reply *struct{}) error {
|
|
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.
|
|
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")
|
|
}
|
|
if args.ID != "" {
|
|
if _, err := uuid.ParseUUID(string(args.ID)); err != nil {
|
|
return fmt.Errorf("Bad node ID: %v", err)
|
|
}
|
|
}
|
|
|
|
// 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.
|
|
func (c *Catalog) Deregister(args *structs.DeregisterRequest, reply *struct{}) error {
|
|
if done, err := c.srv.forward("Catalog.Deregister", args, args, reply); done {
|
|
return err
|
|
}
|
|
defer metrics.MeasureSince([]string{"catalog", "deregister"}, time.Now())
|
|
|
|
// 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)
|
|
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 {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// 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}
|
|
}
|
|
|
|
*reply = dcs
|
|
return nil
|
|
}
|
|
|
|
// ListNodes is used to query the nodes in a DC
|
|
func (c *Catalog) ListNodes(args *structs.DCSpecificRequest, reply *structs.IndexedNodes) error {
|
|
if done, err := c.srv.forward("Catalog.ListNodes", args, args, reply); done {
|
|
return err
|
|
}
|
|
|
|
return c.srv.blockingQuery(
|
|
&args.QueryOptions,
|
|
&reply.QueryMeta,
|
|
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)
|
|
})
|
|
}
|
|
|
|
// ListServices is used to query the services in a DC
|
|
func (c *Catalog) ListServices(args *structs.DCSpecificRequest, reply *structs.IndexedServices) error {
|
|
if done, err := c.srv.forward("Catalog.ListServices", args, args, reply); done {
|
|
return err
|
|
}
|
|
|
|
return c.srv.blockingQuery(
|
|
&args.QueryOptions,
|
|
&reply.QueryMeta,
|
|
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)
|
|
})
|
|
}
|
|
|
|
// ServiceNodes returns all the nodes registered as part of a service
|
|
func (c *Catalog) ServiceNodes(args *structs.ServiceSpecificRequest, reply *structs.IndexedServiceNodes) error {
|
|
if done, err := c.srv.forward("Catalog.ServiceNodes", args, args, reply); done {
|
|
return err
|
|
}
|
|
|
|
// Verify the arguments
|
|
if args.ServiceName == "" && args.ServiceAddress == "" {
|
|
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,
|
|
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
|
|
}
|
|
|
|
// NodeServices returns all the services registered as part of a node
|
|
func (c *Catalog) NodeServices(args *structs.NodeSpecificRequest, reply *structs.IndexedNodeServices) error {
|
|
if done, err := c.srv.forward("Catalog.NodeServices", args, args, reply); done {
|
|
return err
|
|
}
|
|
|
|
// Verify the arguments
|
|
if args.Node == "" {
|
|
return fmt.Errorf("Must provide node")
|
|
}
|
|
|
|
return c.srv.blockingQuery(
|
|
&args.QueryOptions,
|
|
&reply.QueryMeta,
|
|
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)
|
|
})
|
|
}
|