diff --git a/cluster.go b/cluster.go index 140babc11..cc36de430 100644 --- a/cluster.go +++ b/cluster.go @@ -224,6 +224,8 @@ type cluster struct { // nolint: maligned abortAntiEntropyCh chan struct{} + translationSyncer translationSyncer + mu sync.RWMutex jobs map[int64]*resizeJob currentJob *resizeJob @@ -251,6 +253,8 @@ func newCluster() *cluster { closing: make(chan struct{}), joining: make(chan struct{}), + translationSyncer: NopTranslationSyncer, + InternalClient: newNopInternalClient(), logger: logger.NopLogger, @@ -463,7 +467,7 @@ func (c *cluster) unprotectedSetState(state string) { switch state { case ClusterStateNormal, ClusterStateDegraded: - // If state is RESIZING -> NORMAL then run cleanup. + // If state is RESIZING -> [NORMAL, DEGRADED] then run cleanup. if c.state == ClusterStateResizing { doCleanup = true } @@ -471,7 +475,17 @@ func (c *cluster) unprotectedSetState(state string) { c.state = state - if state == ClusterStateResizing { + switch state { + case ClusterStateNormal: + // Because the cluster state is changing to NORMAL, + // we [potentially] need to reset the translation sync. + // If, for example, the cluster has changed size and is + // now settling to NORMAL, the partition ownership may + // have changed, and this will force that to be recalculated. + if err := c.translationSyncer.Reset(); err != nil { + c.logger.Printf("error resetting translation syncer: %s", err) + } + case ClusterStateResizing: c.abortAntiEntropy() } diff --git a/go.sum b/go.sum index 85e0347f6..f932bb2c1 100644 --- a/go.sum +++ b/go.sum @@ -105,6 +105,7 @@ github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic= github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021 h1:ERLyN4p3KS5Fk2ADsDENm2cq0+Lx6sF1sG8uwRlySpU= github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I= +github.com/pilosa/pilosa v1.4.0 h1:nqHNIK4nDslFnem3yDp9R+6TgLdlkY9WdJD88Z83T8U= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= diff --git a/holder.go b/holder.go index e7577f8ed..655182a74 100644 --- a/holder.go +++ b/holder.go @@ -85,6 +85,8 @@ type Holder struct { OpenTranslateStore OpenTranslateStoreFunc OpenTranslateReader OpenTranslateReaderFunc + translationSyncer translationSyncer + // Queue of fields (having a foreign index) which have // opened before their foreign index has opened. foreignIndexFields []*Field @@ -140,6 +142,8 @@ func NewHolder(partitionN int) *Holder { OpenTranslateStore: OpenInMemTranslateStore, + translationSyncer: NopTranslationSyncer, + Logger: logger.NopLogger, } } @@ -228,7 +232,10 @@ func (h *Holder) Open() error { h.snapshotQueue.ScanHolder(h) h.opened.Close() - return nil + + // Since the holder is just opening, this is not so much a reset, + // as it is an initial setting of the transation sync process. + return h.translationSyncer.Reset() } // checkForeignIndex is a check before applying a foreign @@ -484,6 +491,12 @@ func (h *Holder) createIndex(name string, opt IndexOptions) (*Index, error) { // Update options. h.indexes[index.Name()] = index + // Since this is a new index, we need to kick off + // its translation sync. + if err := h.translationSyncer.Reset(); err != nil { + return nil, errors.Wrap(err, "resetting translation sync") + } + return index, nil } @@ -499,6 +512,7 @@ func (h *Holder) newIndex(path, name string) (*Index, error) { index.columnAttrs = h.NewAttrStore(filepath.Join(index.path, ".data")) index.snapshotQueue = h.snapshotQueue index.OpenTranslateStore = h.OpenTranslateStore + index.translationSyncer = h.translationSyncer index.holder = h return index, nil } @@ -527,7 +541,10 @@ func (h *Holder) DeleteIndex(name string) error { // Remove reference. delete(h.indexes, name) - return nil + // I'm not sure if calling Reset() here is necessary + // since closing the index stops its translation + // sync processes. + return h.translationSyncer.Reset() } // Field returns the field for an index and name. @@ -922,8 +939,8 @@ func (s *holderSyncer) syncFragment(index, field, view string, shard uint64) err return nil } -// ResetTranslationSync reinitializes streaming sync of translation data. -func (s *holderSyncer) ResetTranslationSync() error { +// resetTranslationSync reinitializes streaming sync of translation data. +func (s *holderSyncer) resetTranslationSync() error { // Stop existing streams. if err := s.stopTranslationSync(); err != nil { return errors.Wrap(err, "stop translation sync") @@ -944,6 +961,53 @@ func (s *holderSyncer) ResetTranslationSync() error { return nil } +//////////////////////////////////////////////////////////// + +// translationSyncer provides an interface allowing a function +// to notify the server that an action has occurred which requires +// the translation sync process to be reset. In general, this +// includes anything which modifies schema (add/remove index, etc), +// or anything that changes the cluster topology (add/remove node). +// I originally considered leveraging the broadcaster since that was +// already in place and provides similar event messages, but the +// broadcaster is really meant for notifiying other nodes, while +// this is more akin to an internal message bus. In fact, I think +// a future iteration on this may be to make it more generic so +// it can act as an internal message bus where one of the messages +// being published is "translationSyncReset". +type translationSyncer interface { + Reset() error +} + +// NopTranslationSyncer represents a translationSyncer that doesn't do anything. +var NopTranslationSyncer translationSyncer = &nopTranslationSyncer{} + +type nopTranslationSyncer struct{} + +// Reset is a no-op implementation of translationSyncer Reset method. +func (nopTranslationSyncer) Reset() error { return nil } + +// activeTranslationSyncer represents a translationSyncer that resets +// the server's translation syncer. +type activeTranslationSyncer struct { + ch chan struct{} +} + +// newActiveTranslationSyncer returns a new instance of activeTranslationSyncer. +func newActiveTranslationSyncer(ch chan struct{}) *activeTranslationSyncer { + return &activeTranslationSyncer{ + ch: ch, + } +} + +// Reset resets the server's translation syncer. +func (a *activeTranslationSyncer) Reset() error { + a.ch <- struct{}{} + return nil +} + +//////////////////////////////////////////////////////////// + // stopTranslationSync closes and waits for all outstanding translation readers // to complete. This should be called before reconnecting to the cluster in case // of a cluster resize or schema change. @@ -958,17 +1022,36 @@ func (s *holderSyncer) stopTranslationSync() error { return g.Wait() } -// setTranslateReadOnlyFlags updates all translation stores to enabled or disable +// setTranslateReadOnlyFlags updates all translation stores to enable or disable // writing new translation keys. Index stores are writable if the node owns the // partition. Field stores are writable if the node is the coordinator. func (s *holderSyncer) setTranslateReadOnlyFlags() { isCoordinator := s.Cluster.isCoordinator() for _, index := range s.Holder.Indexes() { + // There is a race condition here: + // if Indexes() returns idx1, and then in another + // process, holder.DeleteIndex(idx1) is called, + // then the next step trying to get TranslateStore(partitionID) + // for an index that is closed (and therefore its transateStores + // no longer exist) will fail with a nil pointer error. + // For now, I just checked that the translateStore hasn't been + // set to nil before trying to use it, but another option may + // be to prevent the translateStores from being zeroed out + // while this process is active. Checking for nil as we do + // really obviates the need for the RLock around the for loop. + + // Obtain a read lock on index to prevent Index.Close() from + // destroying the Index.translateStores map before this is + // done using it. + index.mu.RLock() for partitionID := 0; partitionID < s.Cluster.partitionN; partitionID++ { ownsPartition := s.Cluster.ownsPartition(s.Node.ID, partitionID) - index.TranslateStore(partitionID).SetReadOnly(!ownsPartition) + if ts := index.TranslateStore(partitionID); ts != nil { + ts.SetReadOnly(!ownsPartition) + } } + index.mu.RUnlock() for _, field := range index.Fields() { field.TranslateStore().SetReadOnly(!isCoordinator) @@ -1030,16 +1113,13 @@ func (s *holderSyncer) initializeIndexTranslateReplication() error { // initializeFieldTranslateReplication connects the coordinator to stream field data. func (s *holderSyncer) initializeFieldTranslateReplication() error { // Skip if coordinator. - if s.Cluster.Node.IsCoordinator { + if s.Cluster.isCoordinator() { return nil } // Build a map of partition offsets to stream from. m := make(TranslateOffsetMap) for _, index := range s.Holder.Indexes() { - if !index.Keys() { - continue - } for _, field := range index.Fields() { store := field.TranslateStore() offset, err := store.MaxID() diff --git a/index.go b/index.go index b3d97a61f..e0ec7fc3b 100644 --- a/index.go +++ b/index.go @@ -68,6 +68,8 @@ type Index struct { // Per-partition translation stores translateStores map[int]TranslateStore + translationSyncer translationSyncer + // Instantiates new translation stores OpenTranslateStore OpenTranslateStoreFunc } @@ -95,6 +97,8 @@ func NewIndex(path, name string, partitionN int) (*Index, error) { translateStores: make(map[int]TranslateStore), + translationSyncer: NopTranslationSyncer, + OpenTranslateStore: OpenInMemTranslateStore, }, nil } @@ -478,6 +482,11 @@ func (i *Index) createField(name string, opt *FieldOptions) (*Field, error) { // Add to index's field lookup. i.fields[name] = f + // Kick off the field's translation sync process. + if err := i.translationSyncer.Reset(); err != nil { + return nil, errors.Wrap(err, "resetting translation syncer") + } + return f, nil } @@ -533,7 +542,7 @@ func (i *Index) DeleteField(name string) error { // Remove reference. delete(i.fields, name) - return nil + return i.translationSyncer.Reset() } type indexSlice []*Index diff --git a/server.go b/server.go index 5c97dd82c..48ab7bab6 100644 --- a/server.go +++ b/server.go @@ -77,6 +77,9 @@ type Server struct { // nolint: maligned isCoordinator bool syncer holderSyncer + translationSyncer translationSyncer + resetTranslationSyncCh chan struct{} + defaultClient InternalClient dataDir string } @@ -316,10 +319,16 @@ func NewServer(opts ...ServerOption) (*Server, error) { metricInterval: 0, diagnosticInterval: 0, + resetTranslationSyncCh: make(chan struct{}), + logger: logger.NopLogger, } s.cluster.InternalClient = s.defaultClient + s.translationSyncer = newActiveTranslationSyncer(s.resetTranslationSyncCh) + s.holder.translationSyncer = s.translationSyncer + s.cluster.translationSyncer = s.translationSyncer + s.diagnostics.server = s for _, opt := range opts { @@ -503,6 +512,18 @@ func (s *Server) Open() error { log.Println(errors.Wrap(err, "logging startup")) } + // Set up the holderSyncer. + s.syncer.Holder = s.holder + s.syncer.Node = s.cluster.Node + s.syncer.Cluster = s.cluster + s.syncer.Closing = s.closing + s.syncer.Stats = s.holder.Stats.WithTags("HolderSyncer") + + // Start background process listening for translation + // sync resets. + s.wg.Add(1) + go func() { defer s.wg.Done(); s.monitorResetTranslationSync() }() + // Open Cluster management. if err := s.cluster.waitForStarted(); err != nil { return errors.Wrap(err, "opening Cluster") @@ -523,16 +544,6 @@ func (s *Server) Open() error { // buffered channel. s.cluster.listenForJoins() - s.syncer.Holder = s.holder - s.syncer.Node = s.cluster.Node - s.syncer.Cluster = s.cluster - s.syncer.Closing = s.closing - s.syncer.Stats = s.holder.Stats.WithTags("HolderSyncer") - - if err := s.syncer.ResetTranslationSync(); err != nil { - return err - } - // Start background monitoring. s.wg.Add(3) go func() { defer s.wg.Done(); s.monitorAntiEntropy() }() @@ -595,6 +606,34 @@ func (s *Server) SyncData() error { return errors.Wrap(s.syncer.SyncHolder(), "syncing holder") } +// monitorResetTranslationSync is a background process which +// listens for events indicating the need to reset the translation +// sync processes. +func (s *Server) monitorResetTranslationSync() { + s.logger.Printf("holder translation sync monitor initializing") + for { + // Wait for a reset or a close. + select { + case <-s.closing: + return + case <-s.resetTranslationSyncCh: + s.logger.Printf("holder translation sync beginning") + s.wg.Add(1) + go func() { + // Obtaining this lock ensures that there is only + // once instance of resetTranslationSync() running + // at once. + s.syncer.mu.Lock() + defer s.syncer.mu.Unlock() + defer s.wg.Done() + if err := s.syncer.resetTranslationSync(); err != nil { + s.logger.Printf("holder translation sync error: err=%s", err) + } + }() + } + } +} + func (s *Server) monitorAntiEntropy() { if s.antiEntropyInterval == 0 || s.cluster.ReplicaN <= 1 { return // anti entropy disabled @@ -666,16 +705,10 @@ func (s *Server) receiveMessage(m Message) error { if err != nil { return err } - if err := s.syncer.ResetTranslationSync(); err != nil { - return err - } case *DeleteIndexMessage: if err := s.holder.DeleteIndex(obj.Index); err != nil { return err } - if err := s.syncer.ResetTranslationSync(); err != nil { - return err - } case *CreateFieldMessage: idx := s.holder.Index(obj.Index) if idx == nil { @@ -686,17 +719,11 @@ func (s *Server) receiveMessage(m Message) error { if err != nil { return err } - if err := s.syncer.ResetTranslationSync(); err != nil { - return err - } case *DeleteFieldMessage: idx := s.holder.Index(obj.Index) if err := idx.DeleteField(obj.Field); err != nil { return err } - if err := s.syncer.ResetTranslationSync(); err != nil { - return err - } case *DeleteAvailableShardMessage: f := s.holder.Field(obj.Index, obj.Field) if err := f.RemoveAvailableShard(obj.ShardID); err != nil { @@ -725,11 +752,6 @@ func (s *Server) receiveMessage(m Message) error { if err != nil { return err } - if s.syncer.Cluster != nil { - if err := s.syncer.ResetTranslationSync(); err != nil { - return err - } - } case *ResizeInstruction: err := s.cluster.followResizeInstruction(obj) if err != nil {