luxolus
/
open-consul
2
0
Fork 0
Code Issues 1 Pull Requests Packages Releases Wiki Activity

Merge pull request #2661 from hashicorp/f-node-id 

Adds basic support for node IDs.
This commit is contained in:
James Phillips 2017-01-18 16:02:38 -08:00 committed by GitHub
parent 80d3149de2 2e67edb2f2
commit 9e8952c126
27 changed files with 351 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
api/agent_test.go
View File

@ -657,7 +657,7 @@ func TestAgent_Monitor(t *testing.T) {
// Wait for the first log message and validate it
select {
case log := <-logCh:
if !strings.Contains(log, "[INFO] raft: Initial configuration") {
if !strings.Contains(log, "[INFO]") {
t.Fatalf("bad: %q", log)
}
case <-time.After(10 * time.Second):

5
api/catalog.go
View File

@ -1,6 +1,7 @@
package api
type Node struct {
ID string
Node string
Address string
TaggedAddresses map[string]string
@ -8,6 +9,7 @@ type Node struct {
}
type CatalogService struct {
ID string
Node string
Address string
TaggedAddresses map[string]string
@ -28,6 +30,7 @@ type CatalogNode struct {
}
type CatalogRegistration struct {
ID string
Node string
Address string
TaggedAddresses map[string]string
@ -39,7 +42,7 @@ type CatalogRegistration struct {
type CatalogDeregistration struct {
Node string
Address string
Address string // Obsolete.
Datacenter string
ServiceID string
CheckID string

71
command/agent/agent.go
View File

@ -224,7 +224,6 @@ func Create(config *Config, logOutput io.Writer, logWriter *logger.LogWriter,
shutdownCh: make(chan struct{}),
endpoints: make(map[string]string),
}
if err := agent.resolveTmplAddrs(); err != nil {
return nil, err
}
@ -236,6 +235,12 @@ func Create(config *Config, logOutput io.Writer, logWriter *logger.LogWriter,
}
agent.acls = acls
// Retrieve or generate the node ID before setting up the rest of the
// agent, which depends on it.
if err := agent.setupNodeID(config); err != nil {
return nil, fmt.Errorf("Failed to setup node ID: %v", err)
}
// Initialize the local state.
agent.state.Init(config, agent.logger)
@ -303,6 +308,9 @@ func (a *Agent) consulConfig() *consul.Config {
base = consul.DefaultConfig()
}
// This is set when the agent starts up
base.NodeID = a.config.NodeID
// Apply dev mode
base.DevMode = a.config.DevMode
@ -600,6 +608,67 @@ func (a *Agent) setupClient() error {
return nil
}
// setupNodeID will pull the persisted node ID, if any, or create a random one
// and persist it.
func (a *Agent) setupNodeID(config *Config) error {
// If they've configured a node ID manually then just use that, as
// long as it's valid.
if config.NodeID != "" {
if _, err := uuid.ParseUUID(string(config.NodeID)); err != nil {
return err
}
return nil
}
// For dev mode we have no filesystem access so just make a GUID.
if a.config.DevMode {
id, err := uuid.GenerateUUID()
if err != nil {
return err
}
config.NodeID = types.NodeID(id)
a.logger.Printf("[INFO] agent: Generated unique node ID %q for this agent (will not be persisted in dev mode)", config.NodeID)
return nil
}
// Load saved state, if any. Since a user could edit this, we also
// validate it.
fileID := filepath.Join(config.DataDir, "node-id")
if _, err := os.Stat(fileID); err == nil {
rawID, err := ioutil.ReadFile(fileID)
if err != nil {
return err
}
nodeID := strings.TrimSpace(string(rawID))
if _, err := uuid.ParseUUID(nodeID); err != nil {
return err
}
config.NodeID = types.NodeID(nodeID)
}
// If we still don't have a valid node ID, make one.
if config.NodeID == "" {
id, err := uuid.GenerateUUID()
if err != nil {
return err
}
if err := lib.EnsurePath(fileID, false); err != nil {
return err
}
if err := ioutil.WriteFile(fileID, []byte(id), 0600); err != nil {
return err
}
config.NodeID = types.NodeID(id)
a.logger.Printf("[INFO] agent: Generated unique node ID %q for this agent (persisted)", config.NodeID)
}
return nil
}
// setupKeyrings is used to initialize and load keyrings during agent startup
func (a *Agent) setupKeyrings(config *consul.Config) error {
fileLAN := filepath.Join(a.config.DataDir, serfLANKeyring)

67
command/agent/agent_test.go
View File

@ -18,6 +18,8 @@ import (
"github.com/hashicorp/consul/consul/structs"
"github.com/hashicorp/consul/logger"
"github.com/hashicorp/consul/testutil"
"github.com/hashicorp/consul/types"
"github.com/hashicorp/go-uuid"
"github.com/hashicorp/raft"
"strings"
)
@ -308,6 +310,71 @@ func TestAgent_ReconnectConfigSettings(t *testing.T) {
}()
}
func TestAgent_NodeID(t *testing.T) {
c := nextConfig()
dir, agent := makeAgent(t, c)
defer os.RemoveAll(dir)
defer agent.Shutdown()
// The auto-assigned ID should be valid.
id := agent.consulConfig().NodeID
if _, err := uuid.ParseUUID(string(id)); err != nil {
t.Fatalf("err: %v", err)
}
// Running again should get the same ID (persisted in the file).
c.NodeID = ""
if err := agent.setupNodeID(c); err != nil {
t.Fatalf("err: %v", err)
}
if newID := agent.consulConfig().NodeID; id != newID {
t.Fatalf("bad: %q vs %q", id, newID)
}
// Set an invalid ID via config.
c.NodeID = types.NodeID("nope")
err := agent.setupNodeID(c)
if err == nil || !strings.Contains(err.Error(), "uuid string is wrong length") {
t.Fatalf("err: %v", err)
}
// Set a valid ID via config.
newID, err := uuid.GenerateUUID()
if err != nil {
t.Fatalf("err: %v", err)
}
c.NodeID = types.NodeID(newID)
if err := agent.setupNodeID(c); err != nil {
t.Fatalf("err: %v", err)
}
if id := agent.consulConfig().NodeID; string(id) != newID {
t.Fatalf("bad: %q vs. %q", id, newID)
}
// Set an invalid ID via the file.
fileID := filepath.Join(c.DataDir, "node-id")
if err := ioutil.WriteFile(fileID, []byte("adf4238a!882b!9ddc!4a9d!5b6758e4159e"), 0600); err != nil {
t.Fatalf("err: %v", err)
}
c.NodeID = ""
err = agent.setupNodeID(c)
if err == nil || !strings.Contains(err.Error(), "uuid is improperly formatted") {
t.Fatalf("err: %v", err)
}
// Set a valid ID via the file.
if err := ioutil.WriteFile(fileID, []byte("adf4238a-882b-9ddc-4a9d-5b6758e4159e"), 0600); err != nil {
t.Fatalf("err: %v", err)
}
c.NodeID = ""
if err := agent.setupNodeID(c); err != nil {
t.Fatalf("err: %v", err)
}
if id := agent.consulConfig().NodeID; string(id) != "adf4238a-882b-9ddc-4a9d-5b6758e4159e" {
t.Fatalf("bad: %q vs. %q", id, newID)
}
}
func TestAgent_AddService(t *testing.T) {
dir, agent := makeAgent(t, nextConfig())
defer os.RemoveAll(dir)

2
command/agent/command.go
View File

@ -92,6 +92,7 @@ func (c *Command) readConfig() *Config {
cmdFlags.StringVar(&cmdConfig.LogLevel, "log-level", "", "log level")
cmdFlags.StringVar(&cmdConfig.NodeName, "node", "", "node name")
cmdFlags.StringVar((*string)(&cmdConfig.NodeID), "node-id", "", "node ID")
cmdFlags.StringVar(&dcDeprecated, "dc", "", "node datacenter (deprecated: use 'datacenter' instead)")
cmdFlags.StringVar(&cmdConfig.Datacenter, "datacenter", "", "node datacenter")
cmdFlags.StringVar(&cmdConfig.DataDir, "data-dir", "", "path to the data directory")
@ -1115,6 +1116,7 @@ func (c *Command) Run(args []string) int {
c.Ui.Output("Consul agent running!")
c.Ui.Info(fmt.Sprintf(" Version: '%s'", c.HumanVersion))
c.Ui.Info(fmt.Sprintf(" Node ID: '%s'", config.NodeID))
c.Ui.Info(fmt.Sprintf(" Node name: '%s'", config.NodeName))
c.Ui.Info(fmt.Sprintf(" Datacenter: '%s'", config.Datacenter))
c.Ui.Info(fmt.Sprintf(" Server: %v (bootstrap: %v)", config.Server, config.Bootstrap))

8
command/agent/config.go
View File

@ -14,6 +14,7 @@ import (
"github.com/hashicorp/consul/consul"
"github.com/hashicorp/consul/lib"
"github.com/hashicorp/consul/types"
"github.com/hashicorp/consul/watch"
"github.com/mitchellh/mapstructure"
)
@ -312,6 +313,10 @@ type Config struct {
// LogLevel is the level of the logs to putout
LogLevel string `mapstructure:"log_level"`
// Node ID is a unique ID for this node across space and time. Defaults
// to a randomly-generated ID that persists in the data-dir.
NodeID types.NodeID `mapstructure:"node_id"`
// Node name is the name we use to advertise. Defaults to hostname.
NodeName string `mapstructure:"node_name"`
@ -1273,6 +1278,9 @@ func MergeConfig(a, b *Config) *Config {
if b.Protocol > 0 {
result.Protocol = b.Protocol
}
if b.NodeID != "" {
result.NodeID = b.NodeID
}
if b.NodeName != "" {
result.NodeName = b.NodeName
}

8
command/agent/config_test.go
View File

@ -60,7 +60,7 @@ func TestDecodeConfig(t *testing.T) {
}
// Without a protocol
input = `{"node_name": "foo", "datacenter": "dc2"}`
input = `{"node_id": "bar", "node_name": "foo", "datacenter": "dc2"}`
config, err = DecodeConfig(bytes.NewReader([]byte(input)))
if err != nil {
t.Fatalf("err: %s", err)
@ -70,6 +70,10 @@ func TestDecodeConfig(t *testing.T) {
t.Fatalf("bad: %#v", config)
}
if config.NodeID != "bar" {
t.Fatalf("bad: %#v", config)
}
if config.Datacenter != "dc2" {
t.Fatalf("bad: %#v", config)
}
@ -1532,6 +1536,7 @@ func TestMergeConfig(t *testing.T) {
DataDir: "/tmp/foo",
Domain: "basic",
LogLevel: "debug",
NodeID: "bar",
NodeName: "foo",
ClientAddr: "127.0.0.1",
BindAddr: "127.0.0.1",
@ -1586,6 +1591,7 @@ func TestMergeConfig(t *testing.T) {
},
Domain: "other",
LogLevel: "info",
NodeID: "bar",
NodeName: "baz",
ClientAddr: "127.0.0.2",
BindAddr: "127.0.0.2",

6
command/agent/local.go
View File

@ -44,7 +44,7 @@ type localState struct {
iface consul.Interface
// nodeInfoInSync tracks whether the server has our correct top-level
// node information in sync (currently only used for tagged addresses)
// node information in sync
nodeInfoInSync bool
// Services tracks the local services
@ -431,6 +431,7 @@ func (l *localState) setSyncState() error {
// Check the node info
if out1.NodeServices == nil || out1.NodeServices.Node == nil ||
out1.NodeServices.Node.ID != l.config.NodeID ||
!reflect.DeepEqual(out1.NodeServices.Node.TaggedAddresses, l.config.TaggedAddresses) ||
!reflect.DeepEqual(out1.NodeServices.Node.Meta, l.metadata) {
l.nodeInfoInSync = false
@ -633,6 +634,7 @@ func (l *localState) deleteCheck(id types.CheckID) error {
func (l *localState) syncService(id string) error {
req := structs.RegisterRequest{
Datacenter: l.config.Datacenter,
ID: l.config.NodeID,
Node: l.config.NodeName,
Address: l.config.AdvertiseAddr,
TaggedAddresses: l.config.TaggedAddresses,
@ -695,6 +697,7 @@ func (l *localState) syncCheck(id types.CheckID) error {
req := structs.RegisterRequest{
Datacenter: l.config.Datacenter,
ID: l.config.NodeID,
Node: l.config.NodeName,
Address: l.config.AdvertiseAddr,
TaggedAddresses: l.config.TaggedAddresses,
@ -722,6 +725,7 @@ func (l *localState) syncCheck(id types.CheckID) error {
func (l *localState) syncNodeInfo() error {
req := structs.RegisterRequest{
Datacenter: l.config.Datacenter,
ID: l.config.NodeID,
Node: l.config.NodeName,
Address: l.config.AdvertiseAddr,
TaggedAddresses: l.config.TaggedAddresses,

38
command/agent/local_test.go
View File

@ -121,10 +121,14 @@ func TestAgentAntiEntropy_Services(t *testing.T) {
return false, fmt.Errorf("err: %v", err)
}
// Make sure we sent along our tagged addresses when we synced.
// Make sure we sent along our node info when we synced.
id := services.NodeServices.Node.ID
addrs := services.NodeServices.Node.TaggedAddresses
if len(addrs) == 0 || !reflect.DeepEqual(addrs, conf.TaggedAddresses) {
return false, fmt.Errorf("bad: %v", addrs)
meta := services.NodeServices.Node.Meta
if id != conf.NodeID ||
!reflect.DeepEqual(addrs, conf.TaggedAddresses) ||
!reflect.DeepEqual(meta, conf.Meta) {
return false, fmt.Errorf("bad: %v", services.NodeServices.Node)
}
// We should have 6 services (consul included)
@ -717,7 +721,7 @@ func TestAgentAntiEntropy_Checks(t *testing.T) {
}
}
// Make sure we sent along our tagged addresses when we synced.
// Make sure we sent along our node info addresses when we synced.
{
req := structs.NodeSpecificRequest{
Datacenter: "dc1",
@ -728,9 +732,13 @@ func TestAgentAntiEntropy_Checks(t *testing.T) {
t.Fatalf("err: %v", err)
}
id := services.NodeServices.Node.ID
addrs := services.NodeServices.Node.TaggedAddresses
if len(addrs) == 0 || !reflect.DeepEqual(addrs, conf.TaggedAddresses) {
t.Fatalf("bad: %v", addrs)
meta := services.NodeServices.Node.Meta
if id != conf.NodeID ||
!reflect.DeepEqual(addrs, conf.TaggedAddresses) ||
!reflect.DeepEqual(meta, conf.Meta) {
t.Fatalf("bad: %v", services.NodeServices.Node)
}
}
@ -985,6 +993,7 @@ func TestAgentAntiEntropy_Check_DeferSync(t *testing.T) {
func TestAgentAntiEntropy_NodeInfo(t *testing.T) {
conf := nextConfig()
conf.NodeID = types.NodeID("40e4a748-2192-161a-0510-9bf59fe950b5")
conf.Meta["somekey"] = "somevalue"
dir, agent := makeAgent(t, conf)
defer os.RemoveAll(dir)
@ -1020,12 +1029,14 @@ func TestAgentAntiEntropy_NodeInfo(t *testing.T) {
// Make sure we synced our node info - this should have ridden on the
// "consul" service sync
id := services.NodeServices.Node.ID
addrs := services.NodeServices.Node.TaggedAddresses
meta := services.NodeServices.Node.Meta
if len(addrs) == 0 || !reflect.DeepEqual(addrs, conf.TaggedAddresses) || !reflect.DeepEqual(meta, conf.Meta) {
return false, fmt.Errorf("bad: %v", addrs)
if id != conf.NodeID ||
!reflect.DeepEqual(addrs, conf.TaggedAddresses) ||
!reflect.DeepEqual(meta, conf.Meta) {
return false, fmt.Errorf("bad: %v", services.NodeServices.Node)
}
return true, nil
}, func(err error) {
t.Fatalf("err: %s", err)
@ -1045,12 +1056,15 @@ func TestAgentAntiEntropy_NodeInfo(t *testing.T) {
if err := agent.RPC("Catalog.NodeServices", &req, &services); err != nil {
return false, fmt.Errorf("err: %v", err)
}
id := services.NodeServices.Node.ID
addrs := services.NodeServices.Node.TaggedAddresses
meta := services.NodeServices.Node.Meta
if len(addrs) == 0 || !reflect.DeepEqual(addrs, conf.TaggedAddresses) || !reflect.DeepEqual(meta, conf.Meta) {
return false, fmt.Errorf("bad: %v", addrs)
if id != conf.NodeID ||
!reflect.DeepEqual(addrs, conf.TaggedAddresses) ||
!reflect.DeepEqual(meta, conf.Meta) {
return false, fmt.Errorf("bad: %v", services.NodeServices.Node)
}
return true, nil
}, func(err error) {
t.Fatalf("err: %s", err)

6
consul/catalog_endpoint.go
View File

@ -7,6 +7,7 @@ import (
"github.com/armon/go-metrics"
"github.com/hashicorp/consul/consul/structs"
"github.com/hashicorp/consul/types"
"github.com/hashicorp/go-uuid"
)
// Catalog endpoint is used to manipulate the service catalog
@ -25,6 +26,11 @@ func (c *Catalog) Register(args *structs.RegisterRequest, reply *struct{}) error
if args.Node == "" || args.Address == "" {
return fmt.Errorf("Must provide node and address")
}
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.
acl, err := c.srv.resolveToken(args.Token)

38
consul/catalog_endpoint_test.go
View File

@ -11,6 +11,7 @@ import (
"github.com/hashicorp/consul/consul/structs"
"github.com/hashicorp/consul/lib"
"github.com/hashicorp/consul/testutil"
"github.com/hashicorp/consul/types"
"github.com/hashicorp/net-rpc-msgpackrpc"
)
@ -40,13 +41,40 @@ func TestCatalog_Register(t *testing.T) {
if err != nil {
t.Fatalf("err: %v", err)
}
}
testutil.WaitForResult(func() (bool, error) {
err := msgpackrpc.CallWithCodec(codec, "Catalog.Register", &arg, &out)
return err == nil, err
}, func(err error) {
func TestCatalog_Register_NodeID(t *testing.T) {
dir1, s1 := testServer(t)
defer os.RemoveAll(dir1)
defer s1.Shutdown()
codec := rpcClient(t, s1)
defer codec.Close()
arg := structs.RegisterRequest{
Datacenter: "dc1",
ID: "nope",
Node: "foo",
Address: "127.0.0.1",
Service: &structs.NodeService{
Service: "db",
Tags: []string{"master"},
Port: 8000,
},
Check: &structs.HealthCheck{
ServiceID: "db",
},
}
var out struct{}
err := msgpackrpc.CallWithCodec(codec, "Catalog.Register", &arg, &out)
if err == nil || !strings.Contains(err.Error(), "Bad node ID") {
t.Fatalf("err: %v", err)
})
}
arg.ID = types.NodeID("adf4238a-882b-9ddc-4a9d-5b6758e4159e")
if err := msgpackrpc.CallWithCodec(codec, "Catalog.Register", &arg, &out); err != nil {
t.Fatalf("err: %v", err)
}
}
func TestCatalog_Register_ACLDeny(t *testing.T) {

4
consul/client.go
View File

@ -14,6 +14,7 @@ import (
"github.com/hashicorp/consul/consul/agent"
"github.com/hashicorp/consul/consul/servers"
"github.com/hashicorp/consul/consul/structs"
"github.com/hashicorp/consul/lib"
"github.com/hashicorp/serf/coordinate"
"github.com/hashicorp/serf/serf"
)
@ -144,6 +145,7 @@ func (c *Client) setupSerf(conf *serf.Config, ch chan serf.Event, path string) (
conf.NodeName = c.config.NodeName
conf.Tags["role"] = "node"
conf.Tags["dc"] = c.config.Datacenter
conf.Tags["id"] = string(c.config.NodeID)
conf.Tags["vsn"] = fmt.Sprintf("%d", c.config.ProtocolVersion)
conf.Tags["vsn_min"] = fmt.Sprintf("%d", ProtocolVersionMin)
conf.Tags["vsn_max"] = fmt.Sprintf("%d", ProtocolVersionMax)
@ -156,7 +158,7 @@ func (c *Client) setupSerf(conf *serf.Config, ch chan serf.Event, path string) (
conf.RejoinAfterLeave = c.config.RejoinAfterLeave
conf.Merge = &lanMergeDelegate{dc: c.config.Datacenter}
conf.DisableCoordinates = c.config.DisableCoordinates
if err := ensurePath(conf.SnapshotPath, false); err != nil {
if err := lib.EnsurePath(conf.SnapshotPath, false); err != nil {
return nil, err
}
return serf.Create(conf)

4
consul/config.go
View File

@ -8,6 +8,7 @@ import (
"time"
"github.com/hashicorp/consul/tlsutil"
"github.com/hashicorp/consul/types"
"github.com/hashicorp/memberlist"
"github.com/hashicorp/raft"
"github.com/hashicorp/serf/serf"
@ -66,6 +67,9 @@ type Config struct {
// DevMode is used to enable a development server mode.
DevMode bool
// NodeID is a unique identifier for this node across space and time.
NodeID types.NodeID
// Node name is the name we use to advertise. Defaults to hostname.
NodeName string

1
consul/leader.go
View File

@ -418,6 +418,7 @@ AFTER_CHECK:
// Register with the catalog
req := structs.RegisterRequest{
Datacenter: s.config.Datacenter,
ID: types.NodeID(member.Tags["id"]),
Node: member.Name,
Address: member.Addr.String(),
Service: service,

13
consul/server.go
View File

@ -20,6 +20,7 @@ import (
"github.com/hashicorp/consul/consul/agent"
"github.com/hashicorp/consul/consul/state"
"github.com/hashicorp/consul/consul/structs"
"github.com/hashicorp/consul/lib"
"github.com/hashicorp/consul/tlsutil"
"github.com/hashicorp/raft"
"github.com/hashicorp/raft-boltdb"
@ -283,6 +284,10 @@ func NewServer(config *Config) (*Server, error) {
}
go s.wanEventHandler()
// Start monitoring leadership. This must happen after Serf is set up
// since it can fire events when leadership is obtained.
go s.monitorLeadership()
// Start ACL replication.
if s.IsACLReplicationEnabled() {
go s.runACLReplication()
@ -308,6 +313,7 @@ func (s *Server) setupSerf(conf *serf.Config, ch chan serf.Event, path string, w
}
conf.Tags["role"] = "consul"
conf.Tags["dc"] = s.config.Datacenter
conf.Tags["id"] = string(s.config.NodeID)
conf.Tags["vsn"] = fmt.Sprintf("%d", s.config.ProtocolVersion)
conf.Tags["vsn_min"] = fmt.Sprintf("%d", ProtocolVersionMin)
conf.Tags["vsn_max"] = fmt.Sprintf("%d", ProtocolVersionMax)
@ -337,7 +343,7 @@ func (s *Server) setupSerf(conf *serf.Config, ch chan serf.Event, path string, w
// When enabled, the Serf gossip may just turn off if we are the minority
// node which is rather unexpected.
conf.EnableNameConflictResolution = false
if err := ensurePath(conf.SnapshotPath, false); err != nil {
if err := lib.EnsurePath(conf.SnapshotPath, false); err != nil {
return nil, err
}
@ -390,7 +396,7 @@ func (s *Server) setupRaft() error {
} else {
// Create the base raft path.
path := filepath.Join(s.config.DataDir, raftState)
if err := ensurePath(path, true); err != nil {
if err := lib.EnsurePath(path, true); err != nil {
return err
}
@ -489,9 +495,6 @@ func (s *Server) setupRaft() error {
if err != nil {
return err
}
// Start monitoring leadership.
go s.monitorLeadership()
return nil
}

62
consul/state/catalog.go
View File

@ -72,6 +72,7 @@ func (s *StateStore) ensureRegistrationTxn(tx *memdb.Txn, idx uint64, watches *D
req *structs.RegisterRequest) error {
// Add the node.
node := &structs.Node{
ID: req.ID,
Node: req.Node,
Address: req.Address,
TaggedAddresses: req.TaggedAddresses,
@ -229,12 +230,12 @@ func (s *StateStore) NodesByMeta(filters map[string]string) (uint64, structs.Nod
}
// DeleteNode is used to delete a given node by its ID.
func (s *StateStore) DeleteNode(idx uint64, nodeID string) error {
func (s *StateStore) DeleteNode(idx uint64, nodeName string) error {
tx := s.db.Txn(true)
defer tx.Abort()
// Call the node deletion.
if err := s.deleteNodeTxn(tx, idx, nodeID); err != nil {
if err := s.deleteNodeTxn(tx, idx, nodeName); err != nil {
return err
}
@ -244,9 +245,9 @@ func (s *StateStore) DeleteNode(idx uint64, nodeID string) error {
// deleteNodeTxn is the inner method used for removing a node from
// the store within a given transaction.
func (s *StateStore) deleteNodeTxn(tx *memdb.Txn, idx uint64, nodeID string) error {
func (s *StateStore) deleteNodeTxn(tx *memdb.Txn, idx uint64, nodeName string) error {
// Look up the node.
node, err := tx.First("nodes", "id", nodeID)
node, err := tx.First("nodes", "id", nodeName)
if err != nil {
return fmt.Errorf("node lookup failed: %s", err)
}
@ -259,7 +260,7 @@ func (s *StateStore) deleteNodeTxn(tx *memdb.Txn, idx uint64, nodeID string) err
watches := NewDumbWatchManager(s.tableWatches)
// Delete all services associated with the node and update the service index.
services, err := tx.Get("services", "node", nodeID)
services, err := tx.Get("services", "node", nodeName)
if err != nil {
return fmt.Errorf("failed service lookup: %s", err)
}
@ -270,14 +271,14 @@ func (s *StateStore) deleteNodeTxn(tx *memdb.Txn, idx uint64, nodeID string) err
// Do the delete in a separate loop so we don't trash the iterator.
for _, sid := range sids {
if err := s.deleteServiceTxn(tx, idx, watches, nodeID, sid); err != nil {
if err := s.deleteServiceTxn(tx, idx, watches, nodeName, sid); err != nil {
return err
}
}
// Delete all checks associated with the node. This will invalidate
// sessions as necessary.
checks, err := tx.Get("checks", "node", nodeID)
checks, err := tx.Get("checks", "node", nodeName)
if err != nil {
return fmt.Errorf("failed check lookup: %s", err)
}
@ -288,13 +289,13 @@ func (s *StateStore) deleteNodeTxn(tx *memdb.Txn, idx uint64, nodeID string) err
// Do the delete in a separate loop so we don't trash the iterator.
for _, cid := range cids {
if err := s.deleteCheckTxn(tx, idx, watches, nodeID, cid); err != nil {
if err := s.deleteCheckTxn(tx, idx, watches, nodeName, cid); err != nil {
return err
}
}
// Delete any coordinate associated with this node.
coord, err := tx.First("coordinates", "id", nodeID)
coord, err := tx.First("coordinates", "id", nodeName)
if err != nil {
return fmt.Errorf("failed coordinate lookup: %s", err)
}
@ -317,7 +318,7 @@ func (s *StateStore) deleteNodeTxn(tx *memdb.Txn, idx uint64, nodeID string) err
}
// Invalidate any sessions for this node.
sessions, err := tx.Get("sessions", "node", nodeID)
sessions, err := tx.Get("sessions", "node", nodeName)
if err != nil {
return fmt.Errorf("failed session lookup: %s", err)
}
@ -365,9 +366,8 @@ func (s *StateStore) ensureServiceTxn(tx *memdb.Txn, idx uint64, watches *DumbWa
}
// Create the service node entry and populate the indexes. Note that
// conversion doesn't populate any of the node-specific information
// (Address and TaggedAddresses). That's always populated when we read
// from the state store.
// conversion doesn't populate any of the node-specific information.
// That's always populated when we read from the state store.
entry := svc.ToServiceNode(node)
if existing != nil {
entry.CreateIndex = existing.(*structs.ServiceNode).CreateIndex
@ -590,6 +590,7 @@ func (s *StateStore) parseServiceNodes(tx *memdb.Txn, services structs.ServiceNo
// used by agents to perform address translation if they are
// configured to do that.
node := n.(*structs.Node)
s.ID = node.ID
s.Address = node.Address
s.TaggedAddresses = node.TaggedAddresses
s.NodeMeta = node.Meta
@ -601,7 +602,7 @@ func (s *StateStore) parseServiceNodes(tx *memdb.Txn, services structs.ServiceNo
// NodeService is used to retrieve a specific service associated with the given
// node.
func (s *StateStore) NodeService(nodeID string, serviceID string) (uint64, *structs.NodeService, error) {
func (s *StateStore) NodeService(nodeName string, serviceID string) (uint64, *structs.NodeService, error) {
tx := s.db.Txn(false)
defer tx.Abort()
@ -609,9 +610,9 @@ func (s *StateStore) NodeService(nodeID string, serviceID string) (uint64, *stru
idx := maxIndexTxn(tx, s.getWatchTables("NodeService")...)
// Query the service
service, err := tx.First("services", "id", nodeID, serviceID)
service, err := tx.First("services", "id", nodeName, serviceID)
if err != nil {
return 0, nil, fmt.Errorf("failed querying service for node %q: %s", nodeID, err)
return 0, nil, fmt.Errorf("failed querying service for node %q: %s", nodeName, err)
}
if service != nil {
@ -622,7 +623,7 @@ func (s *StateStore) NodeService(nodeID string, serviceID string) (uint64, *stru
}
// NodeServices is used to query service registrations by node ID.
func (s *StateStore) NodeServices(nodeID string) (uint64, *structs.NodeServices, error) {
func (s *StateStore) NodeServices(nodeName string) (uint64, *structs.NodeServices, error) {
tx := s.db.Txn(false)
defer tx.Abort()
@ -630,7 +631,7 @@ func (s *StateStore) NodeServices(nodeID string) (uint64, *structs.NodeServices,
idx := maxIndexTxn(tx, s.getWatchTables("NodeServices")...)
// Query the node
n, err := tx.First("nodes", "id", nodeID)
n, err := tx.First("nodes", "id", nodeName)
if err != nil {
return 0, nil, fmt.Errorf("node lookup failed: %s", err)
}
@ -640,9 +641,9 @@ func (s *StateStore) NodeServices(nodeID string) (uint64, *structs.NodeServices,
node := n.(*structs.Node)
// Read all of the services
services, err := tx.Get("services", "node", nodeID)
services, err := tx.Get("services", "node", nodeName)
if err != nil {
return 0, nil, fmt.Errorf("failed querying services for node %q: %s", nodeID, err)
return 0, nil, fmt.Errorf("failed querying services for node %q: %s", nodeName, err)
}
// Initialize the node services struct
@ -661,13 +662,13 @@ func (s *StateStore) NodeServices(nodeID string) (uint64, *structs.NodeServices,
}
// DeleteService is used to delete a given service associated with a node.
func (s *StateStore) DeleteService(idx uint64, nodeID, serviceID string) error {
func (s *StateStore) DeleteService(idx uint64, nodeName, serviceID string) error {
tx := s.db.Txn(true)
defer tx.Abort()
// Call the service deletion
watches := NewDumbWatchManager(s.tableWatches)
if err := s.deleteServiceTxn(tx, idx, watches, nodeID, serviceID); err != nil {
if err := s.deleteServiceTxn(tx, idx, watches, nodeName, serviceID); err != nil {
return err
}
@ -678,9 +679,9 @@ func (s *StateStore) DeleteService(idx uint64, nodeID, serviceID string) error {
// deleteServiceTxn is the inner method called to remove a service
// registration within an existing transaction.
func (s *StateStore) deleteServiceTxn(tx *memdb.Txn, idx uint64, watches *DumbWatchManager, nodeID, serviceID string) error {
func (s *StateStore) deleteServiceTxn(tx *memdb.Txn, idx uint64, watches *DumbWatchManager, nodeName, serviceID string) error {
// Look up the service.
service, err := tx.First("services", "id", nodeID, serviceID)
service, err := tx.First("services", "id", nodeName, serviceID)
if err != nil {
return fmt.Errorf("failed service lookup: %s", err)
}
@ -690,7 +691,7 @@ func (s *StateStore) deleteServiceTxn(tx *memdb.Txn, idx uint64, watches *DumbWa
// Delete any checks associated with the service. This will invalidate
// sessions as necessary.
checks, err := tx.Get("checks", "node_service", nodeID, serviceID)
checks, err := tx.Get("checks", "node_service", nodeName, serviceID)
if err != nil {
return fmt.Errorf("failed service check lookup: %s", err)
}
@ -701,7 +702,7 @@ func (s *StateStore) deleteServiceTxn(tx *memdb.Txn, idx uint64, watches *DumbWa
// Do the delete in a separate loop so we don't trash the iterator.
for _, cid := range cids {
if err := s.deleteCheckTxn(tx, idx, watches, nodeID, cid); err != nil {
if err := s.deleteCheckTxn(tx, idx, watches, nodeName, cid); err != nil {
return err
}
}
@ -825,7 +826,7 @@ func (s *StateStore) ensureCheckTxn(tx *memdb.Txn, idx uint64, watches *DumbWatc
// NodeCheck is used to retrieve a specific check associated with the given
// node.
func (s *StateStore) NodeCheck(nodeID string, checkID types.CheckID) (uint64, *structs.HealthCheck, error) {
func (s *StateStore) NodeCheck(nodeName string, checkID types.CheckID) (uint64, *structs.HealthCheck, error) {
tx := s.db.Txn(false)
defer tx.Abort()
@ -833,7 +834,7 @@ func (s *StateStore) NodeCheck(nodeID string, checkID types.CheckID) (uint64, *s
idx := maxIndexTxn(tx, s.getWatchTables("NodeCheck")...)
// Return the check.
check, err := tx.First("checks", "id", nodeID, string(checkID))
check, err := tx.First("checks", "id", nodeName, string(checkID))
if err != nil {
return 0, nil, fmt.Errorf("failed check lookup: %s", err)
}
@ -846,7 +847,7 @@ func (s *StateStore) NodeCheck(nodeID string, checkID types.CheckID) (uint64, *s
// NodeChecks is used to retrieve checks associated with the
// given node from the state store.
func (s *StateStore) NodeChecks(nodeID string) (uint64, structs.HealthChecks, error) {
func (s *StateStore) NodeChecks(nodeName string) (uint64, structs.HealthChecks, error) {
tx := s.db.Txn(false)
defer tx.Abort()
@ -854,7 +855,7 @@ func (s *StateStore) NodeChecks(nodeID string) (uint64, structs.HealthChecks, er
idx := maxIndexTxn(tx, s.getWatchTables("NodeChecks")...)
// Return the checks.
checks, err := tx.Get("checks", "node", nodeID)
checks, err := tx.Get("checks", "node", nodeName)
if err != nil {
return 0, nil, fmt.Errorf("failed check lookup: %s", err)
}
@ -1195,6 +1196,7 @@ func (s *StateStore) parseNodes(tx *memdb.Txn, idx uint64,
// Create the wrapped node
dump := &structs.NodeInfo{
ID: node.ID,
Node: node.Node,
Address: node.Address,
TaggedAddresses: node.TaggedAddresses,

4
consul/state/catalog_test.go
View File

@ -16,6 +16,7 @@ func TestStateStore_EnsureRegistration(t *testing.T) {
// Start with just a node.
req := &structs.RegisterRequest{
ID: types.NodeID("40e4a748-2192-161a-0510-9bf59fe950b5"),
Node: "node1",
Address: "1.2.3.4",
TaggedAddresses: map[string]string{
@ -35,7 +36,8 @@ func TestStateStore_EnsureRegistration(t *testing.T) {
if err != nil {
t.Fatalf("err: %s", err)
}
if out.Node != "node1" || out.Address != "1.2.3.4" ||
if out.ID != types.NodeID("40e4a748-2192-161a-0510-9bf59fe950b5") ||
out.Node != "node1" || out.Address != "1.2.3.4" ||
len(out.TaggedAddresses) != 1 ||
out.TaggedAddresses["hello"] != "world" ||
out.Meta["somekey"] != "somevalue" ||

19
consul/structs/structs.go
View File

@ -170,6 +170,7 @@ type QueryMeta struct {
// is provided, the node is registered.
type RegisterRequest struct {
Datacenter string
ID types.NodeID
Node string
Address string
TaggedAddresses map[string]string
@ -194,7 +195,8 @@ func (r *RegisterRequest) ChangesNode(node *Node) bool {
}
// Check if any of the node-level fields are being changed.
if r.Node != node.Node ||
if r.ID != node.ID ||
r.Node != node.Node ||
r.Address != node.Address ||
!reflect.DeepEqual(r.TaggedAddresses, node.TaggedAddresses) ||
!reflect.DeepEqual(r.NodeMeta, node.Meta) {
@ -280,6 +282,7 @@ func (r *ChecksInStateRequest) RequestDatacenter() string {
// Used to return information about a node
type Node struct {
ID types.NodeID
Node string
Address string
TaggedAddresses map[string]string
@ -303,12 +306,13 @@ func SatisfiesMetaFilters(meta map[string]string, filters map[string]string) boo
// Maps service name to available tags
type Services map[string][]string
// ServiceNode represents a node that is part of a service. Address, TaggedAddresses,
// and NodeMeta are node-related fields that are always empty in the state
// store and are filled in on the way out by parseServiceNodes(). This is also
// why PartialClone() skips them, because we know they are blank already so it
// would be a waste of time to copy them.
// ServiceNode represents a node that is part of a service. ID, Address,
// TaggedAddresses, and NodeMeta are node-related fields that are always empty
// in the state store and are filled in on the way out by parseServiceNodes().
// This is also why PartialClone() skips them, because we know they are blank
// already so it would be a waste of time to copy them.
type ServiceNode struct {
ID types.NodeID
Node string
Address string
TaggedAddresses map[string]string
@ -330,6 +334,7 @@ func (s *ServiceNode) PartialClone() *ServiceNode {
copy(tags, s.ServiceTags)
return &ServiceNode{
// Skip ID, see above.
Node: s.Node,
// Skip Address, see above.
// Skip TaggedAddresses, see above.
@ -396,6 +401,7 @@ func (s *NodeService) IsSame(other *NodeService) bool {
// ToServiceNode converts the given node service to a service node.
func (s *NodeService) ToServiceNode(node string) *ServiceNode {
return &ServiceNode{
// Skip ID, see ServiceNode definition.
Node: node,
// Skip Address, see ServiceNode definition.
// Skip TaggedAddresses, see ServiceNode definition.
@ -502,6 +508,7 @@ OUTER:
// a node. This is currently used for the UI only, as it is
// rather expensive to generate.
type NodeInfo struct {
ID types.NodeID
Node string
Address string
TaggedAddresses map[string]string

11
consul/structs/structs_test.go
View File

@ -107,6 +107,7 @@ func TestStructs_ACL_IsSame(t *testing.T) {
func TestStructs_RegisterRequest_ChangesNode(t *testing.T) {
req := &RegisterRequest{
ID: types.NodeID("40e4a748-2192-161a-0510-9bf59fe950b5"),
Node: "test",
Address: "127.0.0.1",
TaggedAddresses: make(map[string]string),
@ -116,6 +117,7 @@ func TestStructs_RegisterRequest_ChangesNode(t *testing.T) {
}
node := &Node{
ID: types.NodeID("40e4a748-2192-161a-0510-9bf59fe950b5"),
Node: "test",
Address: "127.0.0.1",
TaggedAddresses: make(map[string]string),
@ -140,6 +142,7 @@ func TestStructs_RegisterRequest_ChangesNode(t *testing.T) {
}
}
check(func() { req.ID = "nope" }, func() { req.ID = types.NodeID("40e4a748-2192-161a-0510-9bf59fe950b5") })
check(func() { req.Node = "nope" }, func() { req.Node = "test" })
check(func() { req.Address = "127.0.0.2" }, func() { req.Address = "127.0.0.1" })
check(func() { req.TaggedAddresses["wan"] = "nope" }, func() { delete(req.TaggedAddresses, "wan") })
@ -153,6 +156,7 @@ func TestStructs_RegisterRequest_ChangesNode(t *testing.T) {
// testServiceNode gives a fully filled out ServiceNode instance.
func testServiceNode() *ServiceNode {
return &ServiceNode{
ID: types.NodeID("40e4a748-2192-161a-0510-9bf59fe950b5"),
Node: "node1",
Address: "127.0.0.1",
TaggedAddresses: map[string]string{
@ -182,10 +186,14 @@ func TestStructs_ServiceNode_PartialClone(t *testing.T) {
// Make sure the parts that weren't supposed to be cloned didn't get
// copied over, then zero-value them out so we can do a DeepEqual() on
// the rest of the contents.
if clone.Address != "" || len(clone.TaggedAddresses) != 0 || len(clone.NodeMeta) != 0 {
if clone.ID != "" ||
clone.Address != "" ||
len(clone.TaggedAddresses) != 0 ||
len(clone.NodeMeta) != 0 {
t.Fatalf("bad: %v", clone)
}
sn.ID = ""
sn.Address = ""
sn.TaggedAddresses = nil
sn.NodeMeta = nil
@ -206,6 +214,7 @@ func TestStructs_ServiceNode_Conversions(t *testing.T) {
// These two fields get lost in the conversion, so we have to zero-value
// them out before we do the compare.
sn.ID = ""
sn.Address = ""
sn.TaggedAddresses = nil
sn.NodeMeta = nil

10
consul/util.go
View File

@ -4,8 +4,6 @@ import (
"encoding/binary"
"fmt"
"net"
"os"
"path/filepath"
"runtime"
"strconv"
@ -64,14 +62,6 @@ func init() {
privateBlocks[5] = block
}
// ensurePath is used to make sure a path exists
func ensurePath(path string, dir bool) error {
if !dir {
path = filepath.Dir(path)
}
return os.MkdirAll(path, 0755)
}
// CanServersUnderstandProtocol checks to see if all the servers in the given
// list understand the given protocol version. If there are no servers in the
// list then this will return false.

14
lib/path.go Normal file
View File

@ -0,0 +1,14 @@
package lib
import (
"os"
"path/filepath"
)
// EnsurePath is used to make sure a path exists
func EnsurePath(path string, dir bool) error {
if !dir {
path = filepath.Dir(path)
}
return os.MkdirAll(path, 0755)
}

4
types/node_id.go Normal file
View File

@ -0,0 +1,4 @@
package types
// NodeID is a unique identifier for a node across space and time.
type NodeID string

2
website/source/docs/agent/http/agent.html.markdown
View File

@ -143,6 +143,7 @@ It returns a JSON body like this:
"DNSRecursors": [],
"Domain": "consul.",
"LogLevel": "INFO",
"NodeID": "40e4a748-2192-161a-0510-9bf59fe950b5",
"NodeName": "foobar",
"ClientAddr": "127.0.0.1",
"BindAddr": "0.0.0.0",
@ -183,6 +184,7 @@ It returns a JSON body like this:
"Tags": {
"bootstrap": "1",
"dc": "dc1",
"id": "40e4a748-2192-161a-0510-9bf59fe950b5",
"port": "8300",
"role": "consul",
"vsn": "1",

12
website/source/docs/agent/http/catalog.html.markdown
View File

@ -38,6 +38,7 @@ body must look something like:
```javascript
{
"Datacenter": "dc1",
"ID": "40e4a748-2192-161a-0510-9bf59fe950b5",
"Node": "foobar",
"Address": "192.168.10.10",
"TaggedAddresses": {
@ -74,7 +75,10 @@ to match that of the agent. If only those are provided, the endpoint will regist
the node with the catalog. `TaggedAddresses` can be used in conjunction with the
[`translate_wan_addrs`](/docs/agent/options.html#translate_wan_addrs) configuration
option and the `wan` address. The `lan` address was added in Consul 0.7 to help find
the LAN address if address translation is enabled.
the LAN address if address translation is enabled. The `ID` field was added in Consul
0.7.3 and is optional, but if supplied must be in the form of a hex string, 36
characters long. This is a unique identifier for this node across all time, even if
the node name or address changes.
The `Meta` block was added in Consul 0.7.3 to enable associating arbitrary metadata
key/value pairs with a node for filtering purposes. For more information on node metadata,
@ -208,6 +212,7 @@ It returns a JSON body like this:
```javascript
[
{
"ID": "40e4a748-2192-161a-0510-9bf59fe950b5",
"Node": "baz",
"Address": "10.1.10.11",
"TaggedAddresses": {
@ -219,6 +224,7 @@ It returns a JSON body like this:
}
},
{
"ID": "8f246b77-f3e1-ff88-5b48-8ec93abf3e05",
"Node": "foobar",
"Address": "10.1.10.12",
"TaggedAddresses": {
@ -288,6 +294,8 @@ It returns a JSON body like this:
```javascript
[
{
"ID": "40e4a748-2192-161a-0510-9bf59fe950b5",
"Node": "foobar",
"Address": "192.168.10.10",
"TaggedAddresses": {
"lan": "192.168.10.10",
@ -298,7 +306,6 @@ It returns a JSON body like this:
}
"CreateIndex": 51,
"ModifyIndex": 51,
"Node": "foobar",
"ServiceAddress": "172.17.0.3",
"ServiceEnableTagOverride": false,
"ServiceID": "32a2a47f7992:nodea:5000",
@ -340,6 +347,7 @@ It returns a JSON body like this:
```javascript
{
"Node": {
"ID": "40e4a748-2192-161a-0510-9bf59fe950b5",
"Node": "foobar",
"Address": "10.1.10.12",
"TaggedAddresses": {

3
website/source/docs/agent/http/health.html.markdown
View File

@ -33,6 +33,7 @@ It returns a JSON body like this:
```javascript
[
{
"ID": "40e4a748-2192-161a-0510-9bf59fe950b5",
"Node": "foobar",
"CheckID": "serfHealth",
"Name": "Serf Health Status",
@ -43,6 +44,7 @@ It returns a JSON body like this:
"ServiceName": ""
},
{
"ID": "40e4a748-2192-161a-0510-9bf59fe950b5",
"Node": "foobar",
"CheckID": "service:redis",
"Name": "Service 'redis' check",
@ -136,6 +138,7 @@ It returns a JSON body like this:
[
{
"Node": {
"ID": "40e4a748-2192-161a-0510-9bf59fe950b5",
"Node": "foobar",
"Address": "10.1.10.12",
"TaggedAddresses": {

1
website/source/docs/agent/http/query.html.markdown
View File

@ -402,6 +402,7 @@ a JSON body will be returned like this:
"Nodes": [
{
"Node": {
"ID": "40e4a748-2192-161a-0510-9bf59fe950b5",
"Node": "foobar",
"Address": "10.1.10.12",
"TaggedAddresses": {

14
website/source/docs/agent/options.html.markdown
View File

@ -282,6 +282,17 @@ will exit with an error at startup.
* <a name="_node"></a><a href="#_node">`-node`</a> - The name of this node in the cluster.
This must be unique within the cluster. By default this is the hostname of the machine.
* <a name="_node_id"></a><a href="#_node_id">`-node-id`</a> - Available in Consul 0.7.3 and later, this
is a unique identifier for this node across all time, even if the name of the node or address
changes. This must be in the form of a hex string, 36 characters long, such as
`adf4238a-882b-9ddc-4a9d-5b6758e4159e`. If this isn't supplied, which is the most common case, then
the agent will generate an identifier at startup and persist it in the <a href="#_data_dir">data directory</a>
so that it will remain the same across agent restarts. This is currently only exposed via
<a href="/docs/agent/http/agent.html#agent_self">/v1/agent/self</a>,
<a href="/docs/agent/http/catalog.html">/v1/catalog</a>, and
<a href="/docs/agent/http/health.html">/v1/health</a> endpoints, but future versions of
Consul will use this to better manage cluster changes, especially for Consul servers.
* <a name="_node_meta"></a><a href="#_node_meta">`-node-meta`</a> - Available in Consul 0.7.3 and later,
this specifies an arbitrary metadata key/value pair to associate with the node, of the form `key:value`.
This can be specified multiple times. Node metadata pairs have the following restrictions:
@ -695,6 +706,9 @@ Consul will not enable TLS for the HTTP API unless the `https` port has been ass
* <a name="log_level"></a><a href="#log_level">`log_level`</a> Equivalent to the
[`-log-level` command-line flag](#_log_level).
* <a name="node_id"></a><a href="#node_id">`node_id`</a> Equivalent to the
[`-node-id` command-line flag](#_node_id).
* <a name="node_name"></a><a href="#node_name">`node_name`</a> Equivalent to the
[`-node` command-line flag](#_node).