MCPcopy Create free account
hub / github.com/cortexproject/cortex / mergeWithTime

Method mergeWithTime

pkg/ring/model.go:223–306  ·  view source on GitHub ↗
(mergeable memberlist.Mergeable, localCAS bool, now time.Time)

Source from the content-addressed store, hash-verified

221}
222
223func (d *Desc) mergeWithTime(mergeable memberlist.Mergeable, localCAS bool, now time.Time) (memberlist.Mergeable, error) {
224 if mergeable == nil {
225 return nil, nil
226 }
227
228 other, ok := mergeable.(*Desc)
229 if !ok {
230 // This method only deals with non-nil rings.
231 return nil, fmt.Errorf("expected *ring.Desc, got %T", mergeable)
232 }
233
234 if other == nil {
235 return nil, nil
236 }
237
238 normalizeIngestersMap(other)
239
240 thisIngesterMap := d.Ingesters
241 otherIngesterMap := other.Ingesters
242
243 var updated []string
244 tokensChanged := false
245
246 maxFutureLimit := now.Add(30 * time.Minute).Unix()
247 for name, oing := range otherIngesterMap {
248 if oing.Timestamp > maxFutureLimit {
249 return nil, fmt.Errorf("ingester %s timestamp in the future, expected max of %d, got %d", name, maxFutureLimit, oing.Timestamp)
250 }
251
252 ting := thisIngesterMap[name]
253 // ting.Timestamp will be 0, if there was no such ingester in our version
254 if oing.Timestamp > ting.Timestamp {
255 if !tokensEqual(ting.Tokens, oing.Tokens) {
256 tokensChanged = true
257 }
258 oing.Tokens = append([]uint32(nil), oing.Tokens...) // make a copy of tokens
259 thisIngesterMap[name] = oing
260 updated = append(updated, name)
261 } else if oing.Timestamp == ting.Timestamp && ting.State != LEFT && oing.State == LEFT {
262 // we accept LEFT even if timestamp hasn't changed
263 thisIngesterMap[name] = oing // has no tokens already
264 updated = append(updated, name)
265 }
266 }
267
268 if localCAS {
269 // This breaks commutativity! But we only do it locally, not when gossiping with others.
270 for name, ting := range thisIngesterMap {
271 if _, ok := otherIngesterMap[name]; !ok && ting.State != LEFT {
272 // missing, let's mark our ingester as LEFT
273 ting.State = LEFT
274 ting.Tokens = nil
275 // We are deleting entry "now", and should not keep old timestamp, because there may already be pending
276 // message in the gossip network with newer timestamp (but still older than "now").
277 // Such message would "resurrect" this deleted entry.
278 ting.Timestamp = now.Unix()
279 thisIngesterMap[name] = ting
280

Callers 2

MergeMethod · 0.95
mergeLocalCASFunction · 0.80

Calls 6

normalizeIngestersMapFunction · 0.85
tokensEqualFunction · 0.85
conflictingTokensExistFunction · 0.85
resolveConflictsFunction · 0.85
NewDescFunction · 0.85
AddMethod · 0.45

Tested by 1

mergeLocalCASFunction · 0.64