Merge pull request #1514 from seebs/discoSyncer

CORE-299 check for nil translate store while reading translate entries
2026-09-07 09:05:55 +00:00 · 2021-03-11 13:03:24 +01:00 · 2021-03-11 13:03:24 +01:00 · 95eaab8ffc
commit 95eaab8ffc
parent 5af850c64a 3fde2dcac2
3 changed files with 140 additions and 108 deletions
--- a/holder.go
+++ b/holder.go
@ -1743,16 +1743,11 @@ func (s *holderSyncer) resetTranslationSync() error {
 	// Set read-only flag for all translation stores.
 	s.setTranslateReadOnlyFlags(snap)

-	// Connect to each node that has a primary for which we are a replica.
-	if err := s.initializeIndexTranslateReplication(snap); err != nil {
-		return errors.Wrap(err, "initialize index translate replication")
-	}
-
-	// Connect to primary to stream field data.
-	if err := s.initializeFieldTranslateReplication(snap); err != nil {
-		return errors.Wrap(err, "initialize field translate replication")
+	if err := s.initializeReplication(snap); err != nil {
+		return errors.Wrap(err, "initializing translation replication")
 	}
 	return nil
+
 }

 ////////////////////////////////////////////////////////////
@ -1859,11 +1854,83 @@ func (s *holderSyncer) setTranslateReadOnlyFlags(snap *topology.ClusterSnapshot)
 	s.Cluster.mu.RUnlock()
 }

-// initializeIndexTranslateReplication connects to each node that is the
-// primary for a partition that we are a replica of.
-func (s *holderSyncer) initializeIndexTranslateReplication(snap *topology.ClusterSnapshot) error {
+// initializeReplication builds a map of nodes for which we need to replicate
+// any key translation, whether that's field keys (every node replicates these
+// from the primary) or index keys (only the replica nodes for each partition
+// replicate these from whichever node is primary for that partition).
+func (s *holderSyncer) initializeReplication(snap *topology.ClusterSnapshot) error {
+	nodeMaps := make(map[string]TranslateOffsetMap)
+
+	if snap.ReplicaN > 1 {
+		if err := s.populateIndexReplication(nodeMaps, snap); err != nil {
+			return err
+		}
+	}
+	if err := s.populateFieldReplication(nodeMaps, snap); err != nil {
+		return err
+	}
+
+	for _, node := range snap.Nodes {
+		m := nodeMaps[node.ID]
+		if m.Empty() {
+			continue
+		}
+
+		// Connect to remote node and begin streaming.
+		rd, err := s.Holder.OpenTranslateReader(context.Background(), node.URI.String(), m)
+		if err != nil {
+			return err
+		}
+		s.readers = append(s.readers, rd)
+
+		s.syncers.Go(func() error {
+			defer rd.Close()
+			s.readBothTranslateReader(rd, snap)
+			return nil
+		})
+	}
+	return nil
+}
+
+// populateFieldReplication populates a map from node IDs to TranslateOffsetMaps
+// to record that we need to translate fields which have key translation
+// from the primary node.
+func (s *holderSyncer) populateFieldReplication(nodeMaps map[string]TranslateOffsetMap, snap *topology.ClusterSnapshot) error {
+	// Set up field translation
+	if !snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID) {
+		primaryID := snap.PrimaryFieldTranslationNode().ID
+		// Build a map of field key offsets to stream from.
+		m := nodeMaps[primaryID]
+		if m == nil {
+			m = make(TranslateOffsetMap)
+			nodeMaps[primaryID] = m
+		}
+		for _, index := range s.Holder.Indexes() {
+			for _, field := range index.Fields() {
+				store := field.TranslateStore()
+				// I think right now this is supposed to be impossible;
+				// we use an InMemTranslateStore by default even if
+				// no translate store is being used or attempted.
+				if store == nil {
+					return fmt.Errorf("no translate store for field %q/%q", index.Name(), field.Name())
+				}
+				offset, err := store.MaxID()
+				if err != nil {
+					return errors.Wrapf(err, "cannot determine max id for %q/%q", index.Name(), field.Name())
+				}
+				m.SetFieldOffset(index.Name(), field.Name(), offset)
+			}
+		}
+	}
+	return nil
+}
+
+// populateIndexReplication populates a map of node IDs to TranslateOffsetMaps
+// to record which nodes we need to replicate index key translation for.
+// That means nodes which are the primary for a partition that we're a
+// non-primary replica for.
+func (s *holderSyncer) populateIndexReplication(nodeMaps map[string]TranslateOffsetMap, snap *topology.ClusterSnapshot) error {
 	for _, node := range snap.Nodes {
-		// Skip local node.
 		if node.ID == s.Node.ID {
 			continue
 		}
@ -1883,6 +1950,9 @@ func (s *holderSyncer) initializeIndexTranslateReplication(snap *topology.Cluste
 				}

 				store := index.TranslateStore(partitionID)
+				if store == nil {
+					return fmt.Errorf("no store available for index %q, partition %d", index.Name(), partitionID)
+				}
 				offset, err := store.MaxID()
 				if err != nil {
 					return errors.Wrapf(err, "cannot determine max id for %q", index.Name())
@ -1895,109 +1965,49 @@ func (s *holderSyncer) initializeIndexTranslateReplication(snap *topology.Cluste
 		if len(m) == 0 {
 			continue
 		}
-
-		// Connect to remote node and begin streaming.
-		rd, err := s.Holder.OpenTranslateReader(context.Background(), node.URI.String(), m)
-		if err != nil {
-			return err
-		}
-		s.readers = append(s.readers, rd)
-
-		s.syncers.Go(func() error {
-			defer rd.Close()
-			s.readIndexTranslateReader(rd)
-			return nil
-		})
+		nodeMaps[node.ID] = m
 	}
-
 	return nil
 }

-// initializeFieldTranslateReplication connects the primary to stream field data.
-func (s *holderSyncer) initializeFieldTranslateReplication(snap *topology.ClusterSnapshot) error {
-	// Skip if primary.
-	if snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID) {
-		return nil
-	}
+// readBothTranslateReader reads key translation for field keys or
+// index keys from a remote node. Both field and index keys may be sent,
+// the distinction is that field keys have a non-empty field name.
+func (s *holderSyncer) readBothTranslateReader(rd TranslateEntryReader, snap *topology.ClusterSnapshot) {
+	for {
+		var entry TranslateEntry
+		if err := rd.ReadEntry(&entry); err != nil {
+			s.Holder.Logger.Printf("cannot read translate entry: %s", err)
+			return
+		}

-	// Build a map of partition offsets to stream from.
-	m := make(TranslateOffsetMap)
-	for _, index := range s.Holder.Indexes() {
-		for _, field := range index.Fields() {
-			store := field.TranslateStore()
-			offset, err := store.MaxID()
-			if err != nil {
-				return errors.Wrapf(err, "cannot determine max id for %q/%q", index.Name(), field.Name())
+		var store TranslateStore
+		if entry.Field != "" {
+			// Find appropriate store.
+			f := s.Holder.Field(entry.Index, entry.Field)
+			if f == nil {
+				s.Holder.Logger.Printf("field not found: %s/%s", entry.Index, entry.Field)
+				return
+			}
+			store = f.TranslateStore()
+			if store == nil {
+				s.Holder.Logger.Printf("no translate store suitable for index %q, field %q, key %q", entry.Index, entry.Field, entry.Key)
+				return
+			}
+		} else {
+			// Find appropriate store.
+			idx := s.Holder.Index(entry.Index)
+			if idx == nil {
+				s.Holder.Logger.Printf("index not found: %q", entry.Index)
+				return
+			}
+			store = idx.TranslateStore(snap.KeyToKeyPartition(entry.Index, entry.Key))
+			if store == nil {
+				s.Holder.Logger.Printf("no translate store suitable for index %q, key %q", entry.Index, entry.Key)
+				return
 			}
-			m.SetFieldOffset(index.Name(), field.Name(), offset)
 		}
-	}
-
-	// Skip if no replication required.
-	if len(m) == 0 {
-		return nil
-	}
-
-	// Connect to primary and begin streaming.
-	primary := snap.PrimaryFieldTranslationNode()
-	rd, err := s.Holder.OpenTranslateReader(context.Background(), primary.URI.String(), m)
-	if err != nil {
-		return err
-	}
-	s.readers = append(s.readers, rd)
-
-	s.syncers.Go(func() error {
-		defer rd.Close()
-		s.readFieldTranslateReader(rd)
-		return nil
-	})
-	return nil
-}
-
-func (s *holderSyncer) readIndexTranslateReader(rd TranslateEntryReader) {
-	// Create a snapshot of the cluster to use for node/partition calculations.
-	snap := topology.NewClusterSnapshot(s.Cluster.noder, s.Cluster.Hasher, s.Cluster.ReplicaN)
-
-	for {
-		var entry TranslateEntry
-		if err := rd.ReadEntry(&entry); err != nil {
-			s.Holder.Logger.Printf("cannot read index translate entry: %s", err)
-			return
-		}
-
-		// Find appropriate store.
-		idx := s.Holder.Index(entry.Index)
-		if idx == nil {
-			s.Holder.Logger.Printf("index not found: %q", entry.Index)
-			return
-		}
-
 		// Apply replication to store.
-		store := idx.TranslateStore(snap.KeyToKeyPartition(entry.Index, entry.Key))
-		if err := store.ForceSet(entry.ID, entry.Key); err != nil {
-			s.Holder.Logger.Printf("cannot force set index translation data: %d=%q", entry.ID, entry.Key)
-			return
-		}
-	}
-}
-
-func (s *holderSyncer) readFieldTranslateReader(rd TranslateEntryReader) {
-	for {
-		var entry TranslateEntry
-		if err := rd.ReadEntry(&entry); err != nil {
-			s.Holder.Logger.Printf("cannot read field translate entry: %s", err)
-			return
-		}
-
-		// Find appropriate store.
-		f := s.Holder.Field(entry.Index, entry.Field)
-		if f == nil {
-			s.Holder.Logger.Printf("field not found: %s/%s", entry.Index, entry.Field)
-			return
-		}
-
-		// Apply replication to store.
-		store := f.TranslateStore()
 		if err := store.ForceSet(entry.ID, entry.Key); err != nil {
 			s.Holder.Logger.Printf("cannot force set field translation data: %d=%q", entry.ID, entry.Key)
 			return
--- a/http/translator.go
+++ b/http/translator.go
@ -22,6 +22,7 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"reflect"
 	"sync"

 	"github.com/pilosa/pilosa/v2"
@ -33,8 +34,12 @@ func GetOpenTranslateReaderFunc(client *http.Client) pilosa.OpenTranslateReaderF
 }

 func GetOpenTranslateReaderWithLockerFunc(client *http.Client, locker sync.Locker) pilosa.OpenTranslateReaderFunc {
+	lockType := reflect.TypeOf(locker)
+	if lockType.Kind() == reflect.Ptr {
+		lockType = lockType.Elem()
+	}
 	return func(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap) (pilosa.TranslateEntryReader, error) {
-		return openTranslateReader(ctx, nodeURL, offsets, client, locker)
+		return openTranslateReader(ctx, nodeURL, offsets, client, reflect.New(lockType).Interface().(sync.Locker))
 	}
 }

--- a/translate.go
+++ b/translate.go
@ -304,6 +304,18 @@ func (m TranslateOffsetMap) FieldOffset(index, name string) uint64 {
 	return m[index].Fields[name]
 }

+// Empty reports whether there are any actual entries in the map. This
+// is distinct from len(m) == 0 in that an entry in this map which is
+// itself empty doesn't count as non-empty.
+func (m TranslateOffsetMap) Empty() bool {
+	for _, sub := range m {
+		if !sub.Empty() {
+			return false
+		}
+	}
+	return true
+}
+
 // SetFieldOffset sets the offset for the given field.
 func (m TranslateOffsetMap) SetFieldOffset(index, name string, offset uint64) {
 	if m[index] == nil {
@ -317,6 +329,11 @@ type IndexTranslateOffsetMap struct {
 	Fields     map[string]uint64 `json:"fields"`
 }

+// Empty reports whether this map has neither partitions nor fields.
+func (i *IndexTranslateOffsetMap) Empty() bool {
+	return len(i.Partitions) == 0 && len(i.Fields) == 0
+}
+
 func NewIndexTranslateOffsetMap() *IndexTranslateOffsetMap {
 	return &IndexTranslateOffsetMap{
 		Partitions: make(map[int]uint64),