Merge pull request #868 from molecula/with_primary_instead_owner

Translate only on coordinator/primary

boltdb/translate.go

@@ -210,50 +210,55 @@ func (s *TranslateStore) TranslateKeys(keys []string, writable bool) ([]uint64,
 	return s.translateKeys(keys, writable)
 }
 
-	// Allocate slice for ID mapping.
-	ids = make([]uint64, len(keys))
+func (s *TranslateStore) translateKeys(keys []string, writable bool) ([]uint64, error) {
+	ids := make([]uint64, 0, len(keys))
 
-	// Find ids by key under read lock.
-	var found int
-	if err := s.db.View(func(tx *bolt.Tx) error {
-		bkt := tx.Bucket([]byte("keys"))
-		for i, key := range keys {
-			if id, _ := findIDByKey(bkt, key); id != 0 {
-				ids[i] = id
-				found++
+	if s.ReadOnly() || !writable {
+		found := 0
+		if err := s.db.View(func(tx *bolt.Tx) error {
+			bkt := tx.Bucket(bucketKeys)
+			if bkt == nil {
+				return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys)
 			}
+			for _, key := range keys {
+				if id, _ := findIDByKey(bkt, key); id != 0 {
+					ids = append(ids, id)
+					found++
+				}
+			}
+			return nil
+		}); err != nil {
+			return nil, err
 		}
-		return nil
-	}); err != nil {
-		return nil, err
-	} else if found == len(keys) {
-		return ids, nil
-	}
-
-	if s.ReadOnly() {
-		return ids, pilosa.ErrTranslateStoreReadOnly
+		if found == len(keys) {
+			return ids, nil
+		}
+		if s.ReadOnly() {
+			return ids, pilosa.ErrTranslateStoreReadOnly
+		}
+		if !writable {
+			return nil, pilosa.ErrTranslatingKeyNotFound
+		}
+		return nil, nil
 	}
 
 	// Find or create ids under write lock if any keys were not found.
 	var written bool
 	if err := s.db.Update(func(tx *bolt.Tx) (err error) {
-		bkt := tx.Bucket([]byte("keys"))
-		for i, key := range keys {
-			if ids[i] != 0 {
+		bkt := tx.Bucket(bucketKeys)
+		for _, key := range keys {
+			id, boltKey := findIDByKey(bkt, key)
+			if id != 0 {
+				ids = append(ids, id)
 				continue
 			}
-
-			var boltKey []byte
-			if ids[i], boltKey = findIDByKey(bkt, key); ids[i] != 0 {
-				continue
-			}
-
-			ids[i] = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
-			if err := bkt.Put(boltKey, u64tob(ids[i])); err != nil {
+			id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
+			if err := bkt.Put(boltKey, u64tob(id)); err != nil {
 				return err
-			} else if err := tx.Bucket([]byte("ids")).Put(u64tob(ids[i]), boltKey); err != nil {
+			} else if err := tx.Bucket(bucketIDs).Put(u64tob(id), boltKey); err != nil {
 				return err
 			}
+			ids = append(ids, id)
 			written = true
 		}
 		return nil

cluster.go

@@ -1085,16 +1085,18 @@ func (c *cluster) partitionNodes(partitionID int) []*Node {
 	return nodes
 }
 
-// ownsPartition returns true if a host owns a partition.
-func (c *cluster) ownsPartition(nodeID string, partition int) bool {
+func (c *cluster) primaryPartitionNode(partition int) *Node {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
-	return c.unprotectedOwnsPartition(nodeID, partition)
+	return c.unprotectedPrimaryPartitionNode(partition)
 }
 
-// unprotectedOwnsPartition returns true if a host owns a partition.
-func (c *cluster) unprotectedOwnsPartition(nodeID string, partition int) bool {
-	return Nodes(c.partitionNodes(partition)).ContainsID(nodeID)
+// unprotectedPrimaryPartition returns tprimary node of partition.
+func (c *cluster) unprotectedPrimaryPartitionNode(partition int) *Node {
+	if nodes := c.partitionNodes(partition); len(nodes) > 0 {
+		return nodes[0]
+	}
+	return nil
 }
 
 // containsShards is like OwnsShards, but it includes replicas.
@@ -2335,24 +2337,26 @@ func (c *cluster) translateFieldKey(ctx context.Context, field *Field, key strin
 
 // translateFieldKeys is basically a wrapper around
 // field.TranslateStore().TranslateKey(key), but in
-// the case where the local node's translate store
-// is read-only (i.e. it's not the primary translate
-// store), then this method will forward the translation
+// the case where the local node is not coordinator, then this method will forward the translation
 // request to the coordinator.
-func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys ...string) ([]uint64, error) {
-	ids, err := field.TranslateStore().TranslateKeys(keys)
-	// If we get a "read only" error, then forward the request
-	// to the coordinator.
-	if errors.Cause(err) == ErrTranslateStoreReadOnly {
-		coordinatorNode := c.coordinatorNode()
-
-		ids, err := c.InternalClient.TranslateKeysNode(ctx, &coordinatorNode.URI, field.Index(), field.Name(), keys, writable)
-		if err == nil {
-			return ids, nil
-		}
-		return ids, errors.Wrap(err, "translating field keys on coordinator")
+func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys []string, writable bool) (ids []uint64, err error) {
+	coordinator := c.coordinatorNode()
+	if coordinator == nil {
+		return nil, errors.Errorf("translating field(%s/%s) keys(%v) - cannot find coordinator node", field.Index(), field.Name(), keys)
 	}
-	return ids, err
+
+	if c.Node.ID == coordinator.ID {
+		ids, err = field.TranslateStore().TranslateKeys(keys, writable)
+	} else {
+		// If it's writable, then forward the request to the coordinator.
+		ids, err = c.InternalClient.TranslateKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), keys, writable)
+	}
+
+	if err != nil {
+		return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v)", field.Index(), field.Name(), keys)
+	}
+
+	return ids, nil
 }
 
 func (c *cluster) translateIndexKey(ctx context.Context, indexName string, key string) (uint64, error) {
@@ -2374,11 +2378,18 @@ func (c *cluster) translateIndexKeys(ctx context.Context, indexName string, keys
 		return nil, err
 	}
 
-	ids := make([]uint64, 0, len(keys))
-	for _, k := range keys {
-		if id := keyMap[k]; id != 0 {
-			ids = append(ids, id)
+	// make sure that ids line up with keys, but
+	// not appending, but assigning directly 1:1 into the slice.
+	ids := make([]uint64, len(keys))
+	for i, k := range keys {
+		id, ok := keyMap[k]
+		if !writable {
+			if !ok || id == 0 {
+				c.holder.Logger.Debugf("internal translateIndexKeys error: keyMap had no entry for k='%v', and was not writable", k)
+				return nil, ErrTranslatingKeyNotFound
+			}
 		}
+		ids[i] = id
 	}
 	return ids, nil
 }
@@ -2407,15 +2418,20 @@ func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, ke
 
 		g.Go(func() (err error) {
 			var ids []uint64
-			if c.ownsPartition(c.Node.ID, partitionID) {
-				if ids, err = idx.TranslateStore(partitionID).TranslateKeys(keys); err != nil {
-					return err
-				}
+
+			primary := c.primaryPartitionNode(partitionID)
+			if primary == nil {
+				return errors.Errorf("translating index(%s) keys(%v) on partition(%d) - cannot find primary node", indexName, keys, partitionID)
+			}
+
+			if c.Node.ID == primary.ID {
+				ids, err = idx.TranslateStore(partitionID).TranslateKeys(keys, writable)
 			} else {
-				nodes := c.partitionNodes(partitionID)
-				if ids, err = c.InternalClient.TranslateKeysNode(ctx, &nodes[0].URI, indexName, "", keys, writable); err != nil {
-					return err
-				}
+				ids, err = c.InternalClient.TranslateKeysNode(ctx, &primary.URI, indexName, "", keys, writable)
+			}
+
+			if err != nil {
+				return errors.Wrapf(err, "translating index(%s) keys(%v) on partition(%d)", indexName, keys, partitionID)
 			}
 
 			mu.Lock()
@@ -2476,22 +2492,28 @@ func (c *cluster) translateIndexIDSet(ctx context.Context, indexName string, idS
 
 		g.Go(func() (err error) {
 			var keys []string
-			if c.ownsPartition(c.Node.ID, partitionID) {
-				if keys, err = index.TranslateStore(partitionID).TranslateIDs(ids); err != nil {
-					return err
-				}
+
+			primary := c.primaryPartitionNode(partitionID)
+			if primary == nil {
+				return errors.Errorf("translating index(%s) ids(%v) on partition(%d) - cannot find primary node", indexName, ids, partitionID)
+			}
+
+			if c.Node.ID == primary.ID {
+				keys, err = index.TranslateStore(partitionID).TranslateIDs(ids)
 			} else {
-				nodes := c.partitionNodes(partitionID)
-				if keys, err = c.InternalClient.TranslateIDsNode(ctx, &nodes[0].URI, indexName, "", ids); err != nil {
-					return err
-				}
+				keys, err = c.InternalClient.TranslateIDsNode(ctx, &primary.URI, indexName, "", ids)
+			}
+
+			if err != nil {
+				return errors.Wrapf(err, "translating index(%s) ids(%v) on partition(%d)", indexName, ids, partitionID)
 			}
 
 			mu.Lock()
-			defer mu.Unlock()
-			for i := range ids {
-				idMap[ids[i]] = keys[i]
+			for i, id := range ids {
+				idMap[id] = keys[i]
 			}
+			mu.Unlock()
+
 			return nil
 		})
 	}

holder.go

@@ -1425,9 +1425,11 @@ func (s *holderSyncer) setTranslateReadOnlyFlags() {
 		// done using it.
 		index.mu.RLock()
 		for partitionID := 0; partitionID < s.Cluster.partitionN; partitionID++ {
-			ownsPartition := s.Cluster.unprotectedOwnsPartition(s.Node.ID, partitionID)
+			primary := s.Cluster.unprotectedPrimaryPartitionNode(partitionID)
+			isPrimary := primary != nil && s.Node.ID == primary.ID
+
 			if ts := index.TranslateStore(partitionID); ts != nil {
-				ts.SetReadOnly(!ownsPartition)
+				ts.SetReadOnly(!isPrimary)
 			}
 		}
 		index.mu.RUnlock()

translator_test.go

@@ -330,7 +330,7 @@ func TestTranslation_KeyNotFound(t *testing.T) {
 
 	node0 := c.GetNode(0)
 	node1 := c.GetNode(1)
-	// node2 := c.GetNode(2)
+	node2 := c.GetNode(2)
 	node3 := c.GetNode(3)
 
 	ctx := context.Background()
@@ -362,7 +362,7 @@ func TestTranslation_KeyNotFound(t *testing.T) {
 		if err = node0.API.Serializer.Unmarshal(buf, &resp); err != nil {
 			t.Fatal(err)
 		}
-		id0 := resp.IDs[0]
+		id1 := resp.IDs[0]
 
 		// read non-existing key
 		req, err = node3.API.Serializer.Marshal(&pilosa.TranslateKeysRequest{
@@ -400,23 +400,23 @@ func TestTranslation_KeyNotFound(t *testing.T) {
 			t.Fatalf("TranslateKeys(%+v): expected: nil, got: %d", req, resp)
 		}
 
-		req, err = node1.API.Serializer.Marshal(&pilosa.TranslateKeysRequest{
+		req, err = node2.API.Serializer.Marshal(&pilosa.TranslateKeysRequest{
 			Index:       idx,
 			Field:       fld,
-			Keys:        []string{"k2"},
+			Keys:        []string{"k2", "k1"},
 			NotWritable: false,
 		})
 		if err != nil {
 			t.Fatal(err)
 		}
-		if buf, err = node0.API.TranslateKeys(ctx, bytes.NewReader(req)); err != nil {
+		if buf, err = node2.API.TranslateKeys(ctx, bytes.NewReader(req)); err != nil {
 			t.Fatal(err)
 		}
-		if err = node0.API.Serializer.Unmarshal(buf, &resp); err != nil {
+		if err = node2.API.Serializer.Unmarshal(buf, &resp); err != nil {
 			t.Fatal(err)
 		}
-		if resp.IDs[0] != id0+1 {
-			t.Fatalf("TranslateKeys(%+v): expected: %d, got: %d", req, id0+1, resp.IDs[0])
+		if resp.IDs[0] != id1+1 || resp.IDs[1] != id1 {
+			t.Fatalf("TranslateKeys(%+v): expected: %d,%d, got: %d,%d", req, id1+1, id1, resp.IDs[0], resp.IDs[1])
 		}
 	}
 }