From 1448171a2c1bfdf17267b2d5fe4673b018c58fa6 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Fri, 15 Dec 2017 18:12:28 +0300 Subject: [PATCH] Adds local and cluster IDs --- Gopkg.lock | 8 +++++++- Gopkg.toml | 4 ++++ client.go | 25 +++++++++++++++++++++++++ handler.go | 8 ++++++++ holder.go | 25 +++++++++++++++++++++++++ server.go | 32 +++++++++++++++++++++++++++++++- 6 files changed, 100 insertions(+), 2 deletions(-) diff --git a/Gopkg.lock b/Gopkg.lock index ed772ef39..4114168b3 100644 --- a/Gopkg.lock +++ b/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 diff --git a/Gopkg.toml b/Gopkg.toml index 7ffa67a2f..527744da4 100644 --- a/Gopkg.toml +++ b/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" diff --git a/client.go b/client.go index f71cf57d4..8f4f282f8 100644 --- a/client.go +++ b/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) } diff --git a/handler.go b/handler.go index 7a094bdc9..a5b5c4808 100644 --- a/handler.go +++ b/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{} diff --git a/holder.go b/holder.go index 7cbb8ca42..3f6ccc4fa 100644 --- a/holder.go +++ b/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 { diff --git a/server.go b/server.go index 68e9c6a74..a285d1cca 100644 --- a/server.go +++ b/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 {