diff --git a/boltdb/translate.go b/boltdb/translate.go index e72591abf..8ddcbd555 100644 --- a/boltdb/translate.go +++ b/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 diff --git a/cluster.go b/cluster.go index d0a45f3ee..e4f9aaf18 100644 --- a/cluster.go +++ b/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 }) } diff --git a/holder.go b/holder.go index d1f457371..f394ef7aa 100644 --- a/holder.go +++ b/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() diff --git a/translator_test.go b/translator_test.go index 850eb14b1..86c78bfc6 100644 --- a/translator_test.go +++ b/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]) } } }