memberlist: Use separate queue for CAS messages and only wait for CAS messages queue to be empty when stopping #539
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -157,7 +157,8 @@ type KVConfig struct { | |
| LeftIngestersTimeout time.Duration `yaml:"left_ingesters_timeout" category:"advanced"` | ||
|
|
||
| // Timeout used when leaving the memberlist cluster. | ||
| LeaveTimeout time.Duration `yaml:"leave_timeout" category:"advanced"` | ||
| BroadcastTimeoutForLocalUpdatesOnShutdown time.Duration `yaml:"broadcast_timeout_for_local_updates_on_shutdown" category:"advanced"` | ||
|
|
||
| // How much space to use to keep received and sent messages in memory (for troubleshooting). | ||
| MessageHistoryBufferBytes int `yaml:"message_history_buffer_bytes" category:"advanced"` | ||
|
|
@@ -198,6 +199,7 @@ func (cfg *KVConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix string) { | |
| f.IntVar(&cfg.AdvertisePort, prefix+"memberlist.advertise-port", mlDefaults.AdvertisePort, "Gossip port to advertise to other members in the cluster. Used for NAT traversal.") | ||
| f.StringVar(&cfg.ClusterLabel, prefix+"memberlist.cluster-label", mlDefaults.Label, "The cluster label is an optional string to include in outbound packets and gossip streams. Other members in the memberlist cluster will discard any message whose label doesn't match the configured one, unless the 'cluster-label-verification-disabled' configuration option is set to true.") | ||
| f.BoolVar(&cfg.ClusterLabelVerificationDisabled, prefix+"memberlist.cluster-label-verification-disabled", mlDefaults.SkipInboundLabelCheck, "When true, memberlist doesn't verify that inbound packets and gossip streams have the cluster label matching the configured one. This verification should be disabled while rolling out the change to the configured cluster label in a live memberlist cluster.") | ||
| f.DurationVar(&cfg.BroadcastTimeoutForLocalUpdatesOnShutdown, prefix+"memberlist.broadcast-timeout-for-local-updates-on-shutdown", 10*time.Second, "Timeout for broadcasting all remaining locally-generated updates to other nodes when shutting down. Only used if there are nodes left in the memberlist cluster, and only applies to locally-generated updates, not to broadcast messages that are result of incoming gossip updates. 0 = no timeout, wait until all locally-generated updates are sent.") | ||
|
|
||
| cfg.TCPTransport.RegisterFlagsWithPrefix(f, prefix) | ||
| } | ||
|
|
@@ -231,10 +233,11 @@ type KV struct { | |
| // dns discovery provider | ||
| provider DNSProvider | ||
|
|
||
| // Protects access to memberlist and broadcast queues. | ||
| delegateReady atomic.Bool | ||
| memberlist *memberlist.Memberlist | ||
| localBroadcasts *memberlist.TransmitLimitedQueue // queue for messages generated locally | ||
| gossipBroadcasts *memberlist.TransmitLimitedQueue // queue for messages that we forward from other nodes | ||
|
|
||
| // KV Store. | ||
| storeMu sync.Mutex | ||
|
|
@@ -273,7 +276,8 @@ type KV struct { | |
| numberOfPushes prometheus.Counter | ||
| totalSizeOfPulls prometheus.Counter | ||
| totalSizeOfPushes prometheus.Counter | ||
| numberOfGossipMessagesInQueue prometheus.GaugeFunc | ||
| numberOfLocalMessagesInQueue prometheus.GaugeFunc | ||
| totalSizeOfBroadcastMessagesInQueue prometheus.Gauge | ||
| numberOfBroadcastMessagesDropped prometheus.Counter | ||
| casAttempts prometheus.Counter | ||
|
|
@@ -456,7 +460,11 @@ func (m *KV) starting(ctx context.Context) error { | |
| } | ||
| // Finish delegate initialization. | ||
| m.memberlist = list | ||
| m.localBroadcasts = &memberlist.TransmitLimitedQueue{ | ||
| NumNodes: list.NumMembers, | ||
| RetransmitMult: mlCfg.RetransmitMult, | ||
| } | ||
| m.gossipBroadcasts = &memberlist.TransmitLimitedQueue{ | ||
| NumNodes: list.NumMembers, | ||
| RetransmitMult: mlCfg.RetransmitMult, | ||
| } | ||
|
|
@@ -719,20 +727,24 @@ func (m *KV) discoverMembers(ctx context.Context, members []string) []string { | |
| func (m *KV) stopping(_ error) error { | ||
| level.Info(m.logger).Log("msg", "leaving memberlist cluster") | ||
|
|
||
| // Wait until queue with locally-generated messages is empty, but don't wait for too long. | ||
| // Also don't wait if there is just one node left. | ||
| // Note: Once we enter Stopping state, we don't queue more locally-generated messages. | ||
|
|
||
| deadline := time.Now().Add(m.cfg.BroadcastTimeoutForLocalUpdatesOnShutdown) | ||
|
|
||
| msgs := m.localBroadcasts.NumQueued() | ||
| nodes := m.memberlist.NumMembers() | ||
| for msgs > 0 && nodes > 1 && (m.cfg.BroadcastTimeoutForLocalUpdatesOnShutdown <= 0 || time.Now().Before(deadline)) { | ||
| level.Info(m.logger).Log("msg", "waiting for locally-generated broadcast messages to be sent out", "count", msgs, "nodes", nodes) | ||
| time.Sleep(250 * time.Millisecond) | ||
|
|
||
| msgs = m.localBroadcasts.NumQueued() | ||
| nodes = m.memberlist.NumMembers() | ||
| } | ||
|
|
||
| if msgs > 0 { | ||
| level.Warn(m.logger).Log("msg", "locally-generated broadcast messages left the queue", "count", msgs, "nodes", nodes) | ||
| } | ||
|
|
||
| err := m.memberlist.Leave(m.cfg.LeaveTimeout) | ||
|
|
@@ -972,11 +984,7 @@ outer: | |
| m.casSuccesses.Inc() | ||
| m.notifyWatchers(key) | ||
|
|
||
| m.broadcastNewValue(key, change, newver, codec, true) | ||
| } | ||
|
|
||
| return nil | ||
|
|
@@ -1034,7 +1042,12 @@ func (m *KV) trySingleCas(key string, codec codec.Codec, f func(in interface{}) | |
| return change, newver, retry, nil | ||
| } | ||
|
|
||
| func (m *KV) broadcastNewValue(key string, change Mergeable, version uint, codec codec.Codec, locallyGenerated bool) { | ||
| if locallyGenerated && m.State() != services.Running { | ||
| level.Warn(m.logger).Log("msg", "skipped broadcasting of locally-generated update because memberlist KV is shutting down", "key", key) | ||
| return | ||
| } | ||
|
|
||
| data, err := codec.Encode(change) | ||
| if err != nil { | ||
| level.Error(m.logger).Log("msg", "failed to encode change", "key", key, "version", version, "err", err) | ||
|
|
@@ -1058,7 +1071,25 @@ func (m *KV) broadcastNewValue(key string, change Mergeable, version uint, codec | |
| Changes: change.MergeContent(), | ||
| }) | ||
|
|
||
| l := len(pairData) | ||
| b := ringBroadcast{ | ||
| key: key, | ||
| content: change.MergeContent(), | ||
| version: version, | ||
| msg: pairData, | ||
| finished: func(ringBroadcast) { | ||
| m.totalSizeOfBroadcastMessagesInQueue.Sub(float64(l)) | ||
| }, | ||
| logger: m.logger, | ||
| } | ||
|
|
||
| m.totalSizeOfBroadcastMessagesInQueue.Add(float64(l)) | ||
|
|
||
| if locallyGenerated { | ||
| m.localBroadcasts.QueueBroadcast(b) | ||
| } else { | ||
| m.gossipBroadcasts.QueueBroadcast(b) | ||
| } | ||
| } | ||
|
|
||
| // NodeMeta is method from Memberlist Delegate interface | ||
|
|
@@ -1153,7 +1184,7 @@ func (m *KV) processValueUpdate(workerCh <-chan valueUpdate, key string) { | |
| m.notifyWatchers(key) | ||
|
|
||
| // Don't resend original message, but only changes. | ||
| m.broadcastNewValue(key, mod, version, update.codec, false) | ||
| } | ||
|
|
||
| case <-m.shutdown: | ||
|
|
@@ -1163,32 +1194,25 @@ func (m *KV) processValueUpdate(workerCh <-chan valueUpdate, key string) { | |
| } | ||
| } | ||
|
|
||
| // GetBroadcasts is method from Memberlist Delegate interface | ||
| // It returns all pending broadcasts (within the size limit) | ||
| func (m *KV) GetBroadcasts(overhead, limit int) [][]byte { | ||
| if !m.delegateReady.Load() { | ||
| return nil | ||
| } | ||
|
|
||
| // Prioritize locally-generated messages | ||
| msgs := m.localBroadcasts.GetBroadcasts(overhead, limit) | ||
|
|
||
| // Decrease limit for each message we got from locally-generated broadcasts. | ||
| for _, m := range msgs { | ||
| limit -= overhead + len(m) | ||
| } | ||
|
|
||
| if limit > 0 { | ||
| msgs = append(msgs, m.gossipBroadcasts.GetBroadcasts(overhead, limit)...) | ||
| } | ||
| return msgs | ||
| } | ||
|
|
||
| // LocalState is method from Memberlist Delegate interface | ||
|
|
@@ -1335,7 +1359,7 @@ func (m *KV) MergeRemoteState(data []byte, _ bool) { | |
| level.Error(m.logger).Log("msg", "failed to store received value", "key", kvPair.Key, "err", err) | ||
| } else if newver > 0 { | ||
| m.notifyWatchers(kvPair.Key) | ||
| m.broadcastNewValue(kvPair.Key, change, newver, codec, false) | ||
|
There was a problem hiding this comment. These messages we're persisting to the local KV that come from external broadcasts are non-CAS by definition? (Only the local mutations are called CAS?) There was a problem hiding this comment. Correct. Currently only CAS operation can modify KV store. There is also Delete operation in KV client interface, but memberlist implementation doesn't support it yet. But we should implement it eventually, so perhaps it would be better to call it "local updates", instead of "cas updates". WDYT? I think I'll rename it. There was a problem hiding this comment. Sounds good! CAS is an implementation detail and doesn't really explain that it's a local mod. There was a problem hiding this comment. I've done this change in 979c031. I've also updated |
||
| } | ||
| } | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
inlined into
broadcastNewValue