From 2d22ae55ee06bd62578d1363437494bd978c3db9 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Tue, 7 Jun 2016 14:01:13 -0600 Subject: [PATCH] use db in fragment allocation This commit changes the fragment allocation algorithm in the cluster to make use of the `DB` name. This allows each database to use a different slice distribution. Initially, the `frame` was going to be used for allocation, however, this was problematic since queries can span multiple frames so it's impossible to choose a single frame to use. --- client.go | 14 +++++++------- cluster.go | 15 ++++++++------- cluster_test.go | 9 +++++---- executor.go | 14 +++++++------- fragment.go | 4 ++-- handler.go | 27 ++++++++++++++------------- handler_test.go | 12 ++++++------ index.go | 2 +- 8 files changed, 50 insertions(+), 47 deletions(-) diff --git a/client.go b/client.go index 42302bbaa..008dfa4a9 100644 --- a/client.go +++ b/client.go @@ -90,14 +90,14 @@ func (c *Client) Schema() ([]*DBInfo, error) { return rsp.DBs, nil } -// SliceNodes returns a list of nodes that own a slice. -func (c *Client) SliceNodes(slice uint64) ([]*Node, error) { +// FragmentNodes returns a list of nodes that own a slice. +func (c *Client) FragmentNodes(slice uint64) ([]*Node, error) { // Execute request against the host. u := url.URL{ Scheme: "http", Host: c.host, - Path: "/slices/nodes", - RawQuery: (url.Values{"slice": {strconv.FormatUint(slice, 10)}}).Encode(), + Path: "/fragment/nodes", + RawQuery: (url.Values{"db": {"d"}, "slice": {strconv.FormatUint(slice, 10)}}).Encode(), } resp, err := c.HTTPClient.Get(u.String()) if err != nil { @@ -182,7 +182,7 @@ func (c *Client) Import(db, frame string, slice uint64, bits []Bit) error { } // Retrieve a list of nodes that own the slice. - nodes, err := c.SliceNodes(slice) + nodes, err := c.FragmentNodes(slice) if err != nil { return fmt.Errorf("slice nodes: %s", err) @@ -327,7 +327,7 @@ func (c *Client) backupSliceTo(tw *tar.Writer, db, frame string, slice uint64) e // This function tries slice owners until one succeeds. func (c *Client) BackupSlice(db, frame string, slice uint64) (io.ReadCloser, error) { // Retrieve a list of nodes that own the slice. - nodes, err := c.SliceNodes(slice) + nodes, err := c.FragmentNodes(slice) if err != nil { return nil, fmt.Errorf("slice nodes: %s", err) } @@ -418,7 +418,7 @@ func (c *Client) RestoreFrom(r io.Reader, db, frame string) error { // restoreSliceFrom restores a single slice to all owning nodes. func (c *Client) restoreSliceFrom(buf []byte, db, frame string, slice uint64) error { // Retrieve a list of nodes that own the slice. - nodes, err := c.SliceNodes(slice) + nodes, err := c.FragmentNodes(slice) if err != nil { return fmt.Errorf("slice nodes: %s", err) } diff --git a/cluster.go b/cluster.go index a0afddcae..c0f260af5 100644 --- a/cluster.go +++ b/cluster.go @@ -64,24 +64,25 @@ func NewCluster() *Cluster { } // Partition returns the partition that a slice belongs to. -func (c *Cluster) Partition(slice uint64) int { +func (c *Cluster) Partition(db string, slice uint64) int { var buf [8]byte binary.BigEndian.PutUint64(buf[:], slice) // Hash the bytes and mod by partition count. h := fnv.New64a() + h.Write([]byte(db)) h.Write(buf[:]) return int(h.Sum64() % uint64(c.PartitionN)) } -// SliceNodes returns a list of nodes that own a slice. -func (c *Cluster) SliceNodes(slice uint64) []*Node { - return c.PartitionNodes(c.Partition(slice)) +// FragmentNodes returns a list of nodes that own a fragment. +func (c *Cluster) FragmentNodes(db string, slice uint64) []*Node { + return c.PartitionNodes(c.Partition(db, slice)) } -// OwnsSlice returns true if a host owns slice. -func (c *Cluster) OwnsSlice(host string, slice uint64) bool { - return Nodes(c.SliceNodes(slice)).ContainsHost(host) +// OwnsFragment returns true if a host owns a fragment. +func (c *Cluster) OwnsFragment(host string, db string, slice uint64) bool { + return Nodes(c.FragmentNodes(db, slice)).ContainsHost(host) } // PartitionNodes returns a list of nodes that own a partition. diff --git a/cluster_test.go b/cluster_test.go index 038adaa2a..7b47ff2af 100644 --- a/cluster_test.go +++ b/cluster_test.go @@ -36,11 +36,11 @@ func TestCluster_Owners(t *testing.T) { // Ensure the partitioner can assign a fragment to a partition. func TestCluster_Partition(t *testing.T) { - if err := quick.Check(func(slice uint64, partitionN int) bool { + if err := quick.Check(func(db string, slice uint64, partitionN int) bool { c := pilosa.NewCluster() c.PartitionN = partitionN - partitionID := c.Partition(slice) + partitionID := c.Partition(db, slice) if partitionID < 0 || partitionID > partitionN { t.Errorf("partition out of range: slice=%d, p=%d, n=%d", slice, partitionID, partitionN) } @@ -48,8 +48,9 @@ func TestCluster_Partition(t *testing.T) { return true }, &quick.Config{ Values: func(values []reflect.Value, rand *rand.Rand) { - values[0] = reflect.ValueOf(uint64(rand.Uint32())) - values[1] = reflect.ValueOf(rand.Intn(1000) + 1) + values[0], _ = quick.Value(reflect.TypeOf(""), rand) + values[1] = reflect.ValueOf(uint64(rand.Uint32())) + values[2] = reflect.ValueOf(rand.Intn(1000) + 1) }, }); err != nil { t.Fatal(err) diff --git a/executor.go b/executor.go index c5ce6c457..444630e5e 100644 --- a/executor.go +++ b/executor.go @@ -102,7 +102,7 @@ func (e *Executor) executeCall(db string, c pql.Call, slices []uint64, opt *Exec // executeBitmapCall executes a call that returns a bitmap. func (e *Executor) executeBitmapCall(db string, c pql.BitmapCall, slices []uint64, opt *ExecOptions) (*Bitmap, error) { other := NewBitmap() - for node, nodeSlices := range e.slicesByNode(slices) { + for node, nodeSlices := range e.slicesByNode(db, slices) { // Execute locally if the hostname matches. if node.Host == e.Host { for _, slice := range nodeSlices { @@ -183,7 +183,7 @@ func (e *Executor) executeTopN(db string, c *pql.TopN, slices []uint64, opt *Exe func (e *Executor) executeTopNSlices(db string, c *pql.TopN, slices []uint64, opt *ExecOptions) ([]Pair, error) { var results []Pair - for node, nodeSlices := range e.slicesByNode(slices) { + for node, nodeSlices := range e.slicesByNode(db, slices) { // Execute locally if the hostname matches. if node.Host == e.Host { for _, slice := range nodeSlices { @@ -334,7 +334,7 @@ func (e *Executor) executeUnionSlice(db string, c *pql.Union, slice uint64) (*Bi // executeCount executes a count() call. func (e *Executor) executeCount(db string, c *pql.Count, slices []uint64, opt *ExecOptions) (uint64, error) { var n uint64 - for node, nodeSlices := range e.slicesByNode(slices) { + for node, nodeSlices := range e.slicesByNode(db, slices) { // Execute locally if the hostname matches. if node.Host == e.Host { for _, slice := range nodeSlices { @@ -367,7 +367,7 @@ func (e *Executor) executeProfile(db string, c *pql.Profile, opt *ExecOptions) ( func (e *Executor) executeClearBit(db string, c *pql.ClearBit, opt *ExecOptions) (bool, error) { slice := c.ProfileID / SliceWidth ret := false - for _, node := range e.Cluster.SliceNodes(slice) { + for _, node := range e.Cluster.FragmentNodes(db, slice) { // Update locally if host matches. if node.Host == e.Host { f, err := e.Index.CreateFragmentIfNotExists(db, c.Frame, slice) @@ -399,7 +399,7 @@ func (e *Executor) executeSetBit(db string, c *pql.SetBit, opt *ExecOptions) (bo slice := c.ProfileID / SliceWidth ret := false - for _, node := range e.Cluster.SliceNodes(slice) { + for _, node := range e.Cluster.FragmentNodes(db, slice) { // Update locally if host matches. if node.Host == e.Host { f, err := e.Index.CreateFragmentIfNotExists(db, c.Frame, slice) @@ -558,10 +558,10 @@ func (e *Executor) exec(node *Node, db string, q *pql.Query, slices []uint64, op // slicesByNode returns a mapping of nodes to slices. // // NOTE: Currently the only primary node is used. -func (e *Executor) slicesByNode(slices []uint64) map[*Node][]uint64 { +func (e *Executor) slicesByNode(db string, slices []uint64) map[*Node][]uint64 { m := make(map[*Node][]uint64) for _, slice := range slices { - nodes := e.Cluster.SliceNodes(slice) + nodes := e.Cluster.FragmentNodes(db, slice) node := nodes[0] m[node] = append(m[node], slice) diff --git a/fragment.go b/fragment.go index 9142ad573..cc04ed803 100644 --- a/fragment.go +++ b/fragment.go @@ -1171,7 +1171,7 @@ type FragmentSyncer struct { // then merges any blocks which have differences. func (s *FragmentSyncer) SyncFragment() error { // Determine replica set. - nodes := s.Cluster.SliceNodes(s.Fragment.Slice()) + nodes := s.Cluster.FragmentNodes(s.Fragment.DB(), s.Fragment.Slice()) // Create a set of blocks. blockSets := make([][]FragmentBlock, 0, len(nodes)) @@ -1247,7 +1247,7 @@ func (s *FragmentSyncer) syncBlock(id int) error { // Read pairs from each remote block. var pairSets []PairSet var clients []*Client - for _, node := range s.Cluster.SliceNodes(f.Slice()) { + for _, node := range s.Cluster.FragmentNodes(f.DB(), f.Slice()) { if s.Host == node.Host { continue } diff --git a/handler.go b/handler.go index 07a7f7349..f23227e04 100644 --- a/handler.go +++ b/handler.go @@ -89,13 +89,6 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { default: http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } - case "/slices/nodes": - switch r.Method { - case "GET": - h.handleGetSlicesNodes(w, r) - default: - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) - } case "/slices/max": switch r.Method { case "GET": @@ -103,6 +96,13 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { default: http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } + case "/fragment/nodes": + switch r.Method { + case "GET": + h.handleGetFragmentNodes(w, r) + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } case "/fragment/data": switch r.Method { case "GET": @@ -392,7 +392,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { db, frame, slice := req.GetDB(), req.GetFrame(), req.GetSlice() // Validate that this handler owns the slice. - if !h.Cluster.OwnsSlice(h.Host, slice) { + if !h.Cluster.OwnsFragment(h.Host, db, slice) { http.Error(w, "host does not own slice", http.StatusPreconditionFailed) return } @@ -425,9 +425,10 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { w.Write(buf) } -// handleGetSlicesNodes handles /slices/nodes requests. -func (h *Handler) handleGetSlicesNodes(w http.ResponseWriter, r *http.Request) { +// handleGetFragmentNodes handles /fragment/nodes requests. +func (h *Handler) handleGetFragmentNodes(w http.ResponseWriter, r *http.Request) { q := r.URL.Query() + db := q.Get("db") // Read slice parameter. slice, err := strconv.ParseUint(q.Get("slice"), 10, 64) @@ -436,8 +437,8 @@ func (h *Handler) handleGetSlicesNodes(w http.ResponseWriter, r *http.Request) { return } - // Retrieve slice owner nodes. - nodes := h.Cluster.SliceNodes(slice) + // Retrieve fragment owner nodes. + nodes := h.Cluster.FragmentNodes(db, slice) // Write to response. if err := json.NewEncoder(w).Encode(nodes); err != nil { @@ -597,7 +598,7 @@ func (h *Handler) handlePostFrameRestore(w http.ResponseWriter, r *http.Request) // Loop over each slice and import it if this node owns it. for slice := uint64(0); slice <= sliceN; slice++ { // Ignore this slice if we don't own it. - if !h.Cluster.OwnsSlice(h.Host, slice) { + if !h.Cluster.OwnsFragment(h.Host, db, slice) { continue } diff --git a/handler_test.go b/handler_test.go index 66a72f681..f7723f1d3 100644 --- a/handler_test.go +++ b/handler_test.go @@ -172,7 +172,7 @@ func TestHandler_Query_Bitmap_JSON(t *testing.T) { h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query?db=d", strings.NewReader("Bitmap(100)"))) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) - } else if body := w.Body.String(); body != `{"results":[{"attrs":{"a":"b","c":1,"d":true},"bits":[1,3,66,65537]}]}`+"\n" { + } else if body := w.Body.String(); body != `{"results":[{"attrs":{"a":"b","c":1,"d":true},"bits":[1,3,66,2097153]}]}`+"\n" { t.Fatalf("unexpected body: %s", body) } } @@ -204,7 +204,7 @@ func TestHandler_Query_Bitmap_Profiles_JSON(t *testing.T) { h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query?db=d&profiles=true", strings.NewReader("Bitmap(100)"))) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) - } else if body := w.Body.String(); body != `{"results":[{"attrs":{"a":"b","c":1,"d":true},"bits":[1,3,66,65537]}],"profiles":[{"id":3,"attrs":{"x":"y"}},{"id":66,"attrs":{"y":123,"z":false}}]}`+"\n" { + } else if body := w.Body.String(); body != `{"results":[{"attrs":{"a":"b","c":1,"d":true},"bits":[1,3,66,2097153]}],"profiles":[{"id":3,"attrs":{"x":"y"}},{"id":66,"attrs":{"y":123,"z":false}}]}`+"\n" { t.Fatalf("unexpected body: %s", body) } } @@ -473,18 +473,18 @@ func TestHandler_Version(t *testing.T) { } } -// Ensure the handler can return a list of nodes for a slice. -func TestHandler_Slices_Nodes(t *testing.T) { +// Ensure the handler can return a list of nodes for a fragment. +func TestHandler_Fragment_Nodes(t *testing.T) { h := NewHandler() h.Cluster = NewCluster(3) h.Cluster.ReplicaN = 2 w := httptest.NewRecorder() - r := MustNewHTTPRequest("GET", "/slices/nodes?slice=0", nil) + r := MustNewHTTPRequest("GET", "/fragment/nodes?db=X&slice=0", nil) h.ServeHTTP(w, r) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) - } else if w.Body.String() != `[{"host":"host2"},{"host":"host0"}]`+"\n" { + } else if w.Body.String() != `[{"host":"host1"},{"host":"host2"}]`+"\n" { t.Fatalf("unexpected body: %q", w.Body.String()) } } diff --git a/index.go b/index.go index 3cd890968..4ac614c58 100644 --- a/index.go +++ b/index.go @@ -214,7 +214,7 @@ func (s *IndexSyncer) SyncIndex() error { for _, fi := range di.Frames { for slice := uint64(0); slice <= sliceN; slice++ { // Ignore slices that this host doesn't own. - if !s.Cluster.OwnsSlice(s.Host, slice) { + if !s.Cluster.OwnsFragment(s.Host, di.Name, slice) { continue }