Method updatePartitionRing

ring/partition_ring_watcher.go:109–142

(desc *PartitionRingDesc)

}

func (w *PartitionRingWatcher) updatePartitionRing(desc *PartitionRingDesc) error {
	newRing, err := NewPartitionRingWithOptions(*desc, w.opts)
	if err != nil {
		return errors.Wrap(err, "failed to create partition ring from descriptor")
	}
	w.ringMx.Lock()
	oldRing := w.ring
	w.ring = newRing
	w.ringMx.Unlock()

	if w.delegate != nil {
		w.delegate.OnPartitionRingChanged(&oldRing.desc, desc)
	}

	// Update metrics.
	for state, count := range desc.countPartitionsByState() {
		w.numPartitionsGaugeVec.WithLabelValues(state.CleanName()).Set(float64(count))
	}

	// Check partitions whose state change lock status has changed and log them.
	for partitionID, partition := range desc.Partitions {
		state := partition.GetState().CleanName()

		oldPartition, existedBefore := oldRing.desc.Partitions[partitionID]
		if !existedBefore || partition.StateChangeLocked != oldPartition.StateChangeLocked {
			if partition.StateChangeLocked {
				level.Warn(w.logger).Log("msg", "partition state change is locked", "partition_id", partitionID, "partition_state", state)
			} else if existedBefore {
				level.Info(w.logger).Log("msg", "partition state change is unlocked", "partition_id", partitionID, "partition_state", state)
			}
		}
	}
	return nil
}

// PartitionRing returns the most updated snapshot of the PartitionRing. The returned instance
// is immutable and will not be updated if new changes are done to the ring.