Adds local and cluster IDs

Gopkg.lock

@@ -157,6 +157,12 @@
   revision = "fd36b3595eb2ec8da4b8153b107f7ea08504899d"
   version = "v0.1.1"
 
+[[projects]]
+  name = "github.com/satori/go.uuid"
+  packages = ["."]
+  revision = "879c5887cd475cd7864858769793b2ceb0d44feb"
+  version = "v1.1.0"
+
 [[projects]]
   branch = "master"
   name = "github.com/sean-/seed"
@@ -238,6 +244,6 @@
 [solve-meta]
   analyzer-name = "dep"
   analyzer-version = 1
-  inputs-digest = "75badb0bcc3bb356b04af17979e0af61b4b66c5e0a483f09e39cf1f9b5e5de2c"
+  inputs-digest = "210f654a7a072d5751f0814e4d71ef0758dd53b3dc59ed462619396ef8621d81"
   solver-name = "gps-cdcl"
   solver-version = 1

Gopkg.toml

@@ -1,3 +1,7 @@
 # This file intentionally left blank as all needed dependencies are imported by
 # the project and thus tracked by `dep`.
 # See https://github.com/golang/dep/blob/master/docs/Gopkg.toml.md for details.
+
+[[constraint]]
+  name = "github.com/satori/go.uuid"
+  version = "1.1.0"

client.go

@@ -1087,6 +1087,30 @@ func (c *InternalHTTPClient) clientURI(ctx context.Context) *URI {
 	return clientURI
 }
 
+func (c *InternalHTTPClient) NodeID(uri *URI) (string, error) {
+	u := uriPathToURL(uri, "/id")
+	req, err := http.NewRequest("GET", u.String(), nil)
+	resp, err := c.HTTPClient.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("executing http request: %v", err)
+	}
+	defer resp.Body.Close()
+
+	// Read body.
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return "", fmt.Errorf("reading response body: %v", err)
+	}
+
+	// Return error if status is not OK.
+	switch resp.StatusCode {
+	case http.StatusOK: // ok
+	default:
+		return "", fmt.Errorf("unexpected response status code: %d: %s", resp.StatusCode, body)
+	}
+	return string(body), nil
+}
+
 // Bit represents the location of a single bit.
 type Bit struct {
 	RowID     uint64
@@ -1262,4 +1286,5 @@ type InternalClient interface {
 	ColumnAttrDiff(ctx context.Context, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
 	RowAttrDiff(ctx context.Context, index, frame string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
 	SendMessage(ctx context.Context, pb proto.Message) error
+	NodeID(uri *URI) (string, error)
 }

handler.go

@@ -138,6 +138,7 @@ func NewRouter(handler *Handler) *mux.Router {
 	router.HandleFunc("/version", handler.handleGetVersion).Methods("GET")
 	router.HandleFunc("/recalculate-caches", handler.handleRecalculateCaches).Methods("POST")
 	router.HandleFunc("/cluster/message", handler.handlePostClusterMessage).Methods("POST")
+	router.HandleFunc("/id", handler.handleGetID).Methods("GET")
 
 	// TODO: Apply MethodNotAllowed statuses to all endpoints.
 	// Ideally this would be automatic, as described in this (wontfix) ticket:
@@ -2026,4 +2027,11 @@ func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Reques
 	}
 }
 
+func (h *Handler) handleGetID(w http.ResponseWriter, r *http.Request) {
+	_, err := w.Write([]byte(h.Holder.LocalID))
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+	}
+}
+
 type defaultClusterMessageResponse struct{}

holder.go

@@ -19,14 +19,19 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"log"
 	"net/http"
 	"os"
+	"path"
 	"path/filepath"
 	"sort"
+	"strings"
 	"sync"
 	"syscall"
 	"time"
+
+	uuid "github.com/satori/go.uuid"
 )
 
 const (
@@ -59,6 +64,8 @@ type Holder struct {
 	CacheFlushInterval time.Duration
 
 	LogOutput io.Writer
+
+	LocalID string
 }
 
 // NewHolder returns a new instance of Holder.
@@ -426,6 +433,24 @@ func (h *Holder) setFileLimit() {
 
 func (h *Holder) logger() *log.Logger { return log.New(h.LogOutput, "", log.LstdFlags) }
 
+func (h *Holder) loadLocalID() error {
+	idPath := path.Join(h.Path, "ID")
+	localID := ""
+	localIDBytes, err := ioutil.ReadFile(idPath)
+	if err == nil {
+		localID = strings.TrimSpace(string(localIDBytes))
+	} else {
+		u := uuid.NewV4()
+		localID = u.String()
+		err = ioutil.WriteFile(idPath, []byte(localID), 0600)
+		if err != nil {
+			return err
+		}
+	}
+	h.LocalID = localID
+	return nil
+}
+
 // HolderSyncer is an active anti-entropy tool that compares the local holder
 // with a remote holder based on block checksums and resolves differences.
 type HolderSyncer struct {

server.go

@@ -73,6 +73,7 @@ type Server struct {
 	URI         *URI
 	Cluster     *Cluster
 	diagnostics *diagnostics.Diagnostics
+	ClusterID   string
 
 	// Background monitoring intervals.
 	AntiEntropyInterval time.Duration
@@ -206,6 +207,15 @@ func (s *Server) Open() error {
 		}
 	}()
 
+	// load local ID
+	if err := s.Holder.loadLocalID(); err != nil {
+		s.Logger().Println(err)
+	}
+
+	if err := s.loadClusterID(); err != nil {
+		s.Logger().Println(err)
+	}
+
 	// Start background monitoring.
 	s.wg.Add(4)
 	go func() { defer s.wg.Done(); s.monitorAntiEntropy() }()
@@ -569,7 +579,8 @@ func (s *Server) monitorDiagnostics() {
 	s.diagnostics.Set("Cluster", strings.Join(s.Cluster.NodeSetHosts(), ","))
 	s.diagnostics.Set("NumNodes", len(s.Cluster.Nodes))
 	s.diagnostics.Set("NumCPU", runtime.NumCPU())
-	// TODO: unique cluster ID
+	s.diagnostics.Set("LocalID", s.Holder.LocalID)
+	s.diagnostics.Set("ClusterID", s.ClusterID)
 
 	// Flush the diagnostics metrics at startup, then on each tick interval
 	flush := func() {
@@ -664,6 +675,25 @@ func (s *Server) createDefaultClient(remoteClient *http.Client) {
 	s.defaultClient = NewInternalHTTPClientFromURI(nil, remoteClient)
 }
 
+func (s *Server) loadClusterID() error {
+	// If this is the first node in the cluster, set the ClusterID to its ID
+	node0URI, err := s.Cluster.Nodes[0].URI()
+	if err == nil {
+		if s.URI.Equals(node0URI) {
+			s.ClusterID = s.Holder.LocalID
+			return nil
+		}
+	} else {
+		return err
+	}
+	if clusterID, err := s.defaultClient.NodeID(node0URI); err == nil {
+		s.ClusterID = clusterID
+		return nil
+	} else {
+		return err
+	}
+}
+
 // CountOpenFiles on operating systems that support lsof.
 func CountOpenFiles() (int, error) {
 	switch runtime.GOOS {