mirror of https://github.com/grafana/loki
updated to the latest version of dskit (#9920)
**What this PR does / why we need it**:
Updated Loki with the latest version of dskit
**Checklist**
- [ ] Reviewed the
[`CONTRIBUTING.md`](https://github.com/grafana/loki/blob/main/CONTRIBUTING.md)
guide (**required**)
- [ ] Documentation added
- [ ] Tests updated
- [ ] `CHANGELOG.md` updated
- [ ] If the change is worth mentioning in the release notes, add
`add-to-release-notes` label
- [ ] Changes that require user attention or interaction to upgrade are
documented in `docs/sources/upgrading/_index.md`
- [ ] For Helm chart changes bump the Helm chart version in
`production/helm/loki/Chart.yaml` and update
`production/helm/loki/CHANGELOG.md` and
`production/helm/loki/README.md`. [Example
PR](d10549e3ec)
---------
Signed-off-by: Vladyslav Diachenko <vlad.diachenko@grafana.com>
pull/9910/head^2
Vladyslav Diachenko
3 years ago
committed by
GitHub
parent
bad691b509
commit
3243676f4c
@ -0,0 +1,20 @@ |
||||
/* |
||||
Package ring contains types and functions for creating and working with rings. |
||||
|
||||
# Overview |
||||
|
||||
Rings are shared between instances via a key-value store, and are represented by the [Desc] struct, which contains a map of [InstanceDesc] |
||||
structs representing individual instances in the ring. |
||||
|
||||
# Creating a Ring |
||||
|
||||
Two types are available for creating and updating rings: |
||||
|
||||
- [Lifecycler] - A Service that writes to a ring on behalf of a single instance. It's responsible for claiming tokens on the ring and updating the key/value store with heartbeats, state changes, and token changes. This type is the original lifecycler implementation, and [BasicLifecycler] should be used instead for new services. |
||||
- [BasicLifecycler] - A Service that writes to a ring on behalf of a single instance. It's responsible for claiming tokens on the ring, updating the key/value store with heartbeats and state changes, and uses a delegate with event listeners to help with these. This type is general purpose, is used by numerous services, and is meant for building higher level lifecyclers. |
||||
|
||||
# Observing a ring |
||||
|
||||
The [Ring] type is a Service that is primarily used for reading and watching for changes to a ring. |
||||
*/ |
||||
package ring |
||||
@ -0,0 +1,136 @@ |
||||
package ring |
||||
|
||||
import ( |
||||
"fmt" |
||||
"math" |
||||
"strings" |
||||
) |
||||
|
||||
type ringItem interface { |
||||
key() int |
||||
String() string |
||||
} |
||||
|
||||
type ringInstance struct { |
||||
instanceID int |
||||
} |
||||
|
||||
func (ri ringInstance) key() int { |
||||
return ri.instanceID |
||||
} |
||||
|
||||
func (ri ringInstance) String() string { |
||||
return fmt.Sprintf("[instanceID: %d]", ri.instanceID) |
||||
} |
||||
|
||||
type ringToken struct { |
||||
token uint32 |
||||
prevToken uint32 |
||||
} |
||||
|
||||
func (rt ringToken) key() int { |
||||
return int(rt.token) |
||||
} |
||||
|
||||
func (rt ringToken) String() string { |
||||
return fmt.Sprintf("[token: %d, prevToken: %d]", rt.token, rt.prevToken) |
||||
} |
||||
|
||||
type ownershipInfo[T ringItem] struct { |
||||
item T |
||||
ownership float64 |
||||
} |
||||
|
||||
func newRingTokenOwnershipInfo(token, prevToken uint32) ownershipInfo[ringToken] { |
||||
ownership := float64(tokenDistance(prevToken, token)) |
||||
return ownershipInfo[ringToken]{ |
||||
ownership: ownership, |
||||
item: ringToken{ |
||||
token: token, |
||||
prevToken: prevToken, |
||||
}, |
||||
} |
||||
} |
||||
|
||||
func newRingInstanceOwnershipInfo(instanceID int, ownership float64) ownershipInfo[ringInstance] { |
||||
return ownershipInfo[ringInstance]{ |
||||
ownership: ownership, |
||||
item: ringInstance{ |
||||
instanceID: instanceID, |
||||
}, |
||||
} |
||||
} |
||||
|
||||
// ownershipPriorityQueue is a max-heap, i.e., a priority queue
|
||||
// where items with a higher priority will be extracted first.
|
||||
// Namely, items with a higher ownership have a higher priority.
|
||||
// In order to guarantee that 2 instances of ownershipPriorityQueue
|
||||
// with the same items always assign equal priorities to equal items,
|
||||
// in the case of items with equal ownership, we rely on the
|
||||
// order of item ids.
|
||||
type ownershipPriorityQueue[T ringItem] struct { |
||||
items []ownershipInfo[T] |
||||
} |
||||
|
||||
func newPriorityQueue[T ringItem](capacity int) ownershipPriorityQueue[T] { |
||||
return ownershipPriorityQueue[T]{ |
||||
items: make([]ownershipInfo[T], 0, capacity), |
||||
} |
||||
} |
||||
|
||||
func (pq *ownershipPriorityQueue[T]) Len() int { |
||||
return len(pq.items) |
||||
} |
||||
|
||||
func (pq *ownershipPriorityQueue[T]) Swap(i, j int) { |
||||
pq.items[i], pq.items[j] = pq.items[j], pq.items[i] |
||||
} |
||||
|
||||
func (pq *ownershipPriorityQueue[T]) Less(i, j int) bool { |
||||
if pq.items[i].ownership == pq.items[j].ownership { |
||||
// In order to guarantee the stability, i.e., that the same instanceID and zone as input
|
||||
// always generate the same slice of tokens as output, we enforce that by equal ownership
|
||||
// higher priority is determined by the order of ids.
|
||||
return pq.items[i].item.key() > pq.items[j].item.key() |
||||
} |
||||
// We are implementing a max-heap, so we are using > here.
|
||||
// Since we compare float64, NaN values must be placed at the end.
|
||||
return pq.items[i].ownership > pq.items[j].ownership || (math.IsNaN(pq.items[j].ownership) && !math.IsNaN(pq.items[i].ownership)) |
||||
} |
||||
|
||||
// Push implements heap.Push(any). It pushes the element item onto ownershipPriorityQueue.
|
||||
func (pq *ownershipPriorityQueue[T]) Push(item any) { |
||||
ownershipInfo := item.(ownershipInfo[T]) |
||||
pq.items = append(pq.items, ownershipInfo) |
||||
} |
||||
|
||||
// Pop implements heap.Pop(). It removes and returns the element with the highest priority from ownershipPriorityQueue.
|
||||
func (pq *ownershipPriorityQueue[T]) Pop() any { |
||||
n := len(pq.items) |
||||
item := pq.items[n-1] |
||||
pq.items = pq.items[0 : n-1] |
||||
return item |
||||
} |
||||
|
||||
// Peek the returns the element with the highest priority from ownershipPriorityQueue,
|
||||
// but it does not remove it from the latter. Time complexity is O(1).
|
||||
func (pq *ownershipPriorityQueue[T]) Peek() *ownershipInfo[T] { |
||||
if len(pq.items) == 0 { |
||||
return nil |
||||
} |
||||
return &pq.items[0] |
||||
} |
||||
|
||||
func (pq *ownershipPriorityQueue[T]) String() string { |
||||
return fmt.Sprintf("[%s]", strings.Join(mapItems(pq.items, func(item ownershipInfo[T]) string { |
||||
return fmt.Sprintf("%s-ownership: %.3f", item.item, item.ownership) |
||||
}), ",")) |
||||
} |
||||
|
||||
func mapItems[T, V any](in []T, mapItem func(T) V) []V { |
||||
out := make([]V, len(in)) |
||||
for i, v := range in { |
||||
out[i] = mapItem(v) |
||||
} |
||||
return out |
||||
} |
||||
@ -0,0 +1,305 @@ |
||||
package ring |
||||
|
||||
import ( |
||||
"container/heap" |
||||
"fmt" |
||||
"math" |
||||
"regexp" |
||||
"sort" |
||||
"strconv" |
||||
|
||||
"github.com/go-kit/log" |
||||
"github.com/go-kit/log/level" |
||||
|
||||
"golang.org/x/exp/slices" |
||||
) |
||||
|
||||
const ( |
||||
totalTokensCount = math.MaxUint32 + 1 |
||||
optimalTokensPerInstance = 512 |
||||
maxZonesCount = 8 |
||||
) |
||||
|
||||
var ( |
||||
instanceIDRegex = regexp.MustCompile(`^(.*)-(\d+)$`) |
||||
errorBadInstanceIDFormat = func(instanceID string) error { |
||||
return fmt.Errorf("unable to extract instance id from \"%s\"", instanceID) |
||||
} |
||||
errorZoneCountOutOfBound = func(zonesCount int) error { |
||||
return fmt.Errorf("number of zones %d is not correct: it must be greater than 0 and less or equal than %d", zonesCount, maxZonesCount) |
||||
} |
||||
errorZoneNotValid = func(zone string) error { |
||||
return fmt.Errorf("zone %s is not valid", zone) |
||||
} |
||||
errorMultipleOfZonesCount = func(optimalTokenOwnership uint32, token ringToken) error { |
||||
return fmt.Errorf("calculation of a new token between %d and %d with optimal token ownership %d was impossible: optimal token ownership must be a positive multiple of maximal allowed number of zones %d", token.prevToken, token.token, optimalTokenOwnership, maxZonesCount) |
||||
} |
||||
errorLowerAndUpperBoundModulo = func(optimalTokenOwnership uint32, token ringToken) error { |
||||
return fmt.Errorf("calculation of a new token between %d and %d with optimal token ownership %d was impossible: lower and upper bounds must be congruent modulo maximal allowed number of zones %d", token.prevToken, token.token, optimalTokenOwnership, maxZonesCount) |
||||
} |
||||
errorDistanceBetweenTokensNotBigEnough = func(optimalTokenOwnership int, ownership int64, token ringToken) error { |
||||
return fmt.Errorf("calculation of a new token between %d and %d with optimal token ownership %d was impossible: distance between lower and upper bound %d is not big enough", token.prevToken, token.token, optimalTokenOwnership, ownership) |
||||
} |
||||
) |
||||
|
||||
type SpreadMinimizingTokenGenerator struct { |
||||
instanceID int |
||||
zoneID int |
||||
spreadMinimizingZones []string |
||||
logger log.Logger |
||||
} |
||||
|
||||
func NewSpreadMinimizingTokenGenerator(instance, zone string, spreadMinimizingZones []string, logger log.Logger) (*SpreadMinimizingTokenGenerator, error) { |
||||
if len(spreadMinimizingZones) <= 0 || len(spreadMinimizingZones) > maxZonesCount { |
||||
return nil, errorZoneCountOutOfBound(len(spreadMinimizingZones)) |
||||
} |
||||
sortedZones := make([]string, len(spreadMinimizingZones)) |
||||
copy(sortedZones, spreadMinimizingZones) |
||||
if !slices.IsSorted(sortedZones) { |
||||
sort.Strings(sortedZones) |
||||
} |
||||
instanceID, err := parseInstanceID(instance) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
zoneID, err := findZoneID(zone, sortedZones) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
tokenGenerator := &SpreadMinimizingTokenGenerator{ |
||||
instanceID: instanceID, |
||||
zoneID: zoneID, |
||||
spreadMinimizingZones: sortedZones, |
||||
logger: logger, |
||||
} |
||||
return tokenGenerator, nil |
||||
} |
||||
|
||||
func parseInstanceID(instanceID string) (int, error) { |
||||
parts := instanceIDRegex.FindStringSubmatch(instanceID) |
||||
if len(parts) != 3 { |
||||
return 0, errorBadInstanceIDFormat(instanceID) |
||||
} |
||||
return strconv.Atoi(parts[2]) |
||||
} |
||||
|
||||
// findZoneID gets a zone name and a slice of sorted zones,
|
||||
// and return the index of the zone in the slice.
|
||||
func findZoneID(zone string, sortedZones []string) (int, error) { |
||||
index := slices.Index(sortedZones, zone) |
||||
if index < 0 { |
||||
return 0, errorZoneNotValid(zone) |
||||
} |
||||
return index, nil |
||||
} |
||||
|
||||
// generateFirstInstanceTokens calculates a set of tokens that should be assigned to the first instance (with id 0)
|
||||
// of the zone of the underlying instance.
|
||||
func (t *SpreadMinimizingTokenGenerator) generateFirstInstanceTokens() Tokens { |
||||
// In this approach all the tokens from the same zone are equal to each other modulo maxZonesCount.
|
||||
// Therefore, tokenDistance is calculated as a multiple of maxZonesCount, so that we ensure that
|
||||
// the following for loop calculates the actual tokens following the approach's requirement.
|
||||
tokenDistance := (totalTokensCount / optimalTokensPerInstance / maxZonesCount) * maxZonesCount |
||||
tokens := make(Tokens, 0, optimalTokensPerInstance) |
||||
for i := 0; i < optimalTokensPerInstance; i++ { |
||||
token := uint32(i*tokenDistance) + uint32(t.zoneID) |
||||
tokens = append(tokens, token) |
||||
} |
||||
return tokens |
||||
} |
||||
|
||||
// calculateNewToken determines where in the range represented by the given ringToken should a new token be placed
|
||||
// in order to satisfy the constraint represented by the optimalTokenOwnership. This method assumes that:
|
||||
// - ringToken.token % maxZonesCount == ringToken.prevToken % zonesCount
|
||||
// - optimalTokenOwnership % maxZonesCount == 0,
|
||||
// where zonesCount is the number of zones in the ring. The caller of this function must ensure that these assumptions hold.
|
||||
func (t *SpreadMinimizingTokenGenerator) calculateNewToken(token ringToken, optimalTokenOwnership uint32) (uint32, error) { |
||||
if optimalTokenOwnership < maxZonesCount || optimalTokenOwnership%maxZonesCount != 0 { |
||||
return 0, errorMultipleOfZonesCount(optimalTokenOwnership, token) |
||||
} |
||||
if token.prevToken%maxZonesCount != token.token%maxZonesCount { |
||||
return 0, errorLowerAndUpperBoundModulo(optimalTokenOwnership, token) |
||||
} |
||||
ownership := tokenDistance(token.prevToken, token.token) |
||||
if ownership <= int64(optimalTokenOwnership) { |
||||
return 0, errorDistanceBetweenTokensNotBigEnough(int(optimalTokenOwnership), ownership, token) |
||||
} |
||||
// In the present approach tokens of successive zones are immediate successors of the tokens in
|
||||
// the previous zone. This means that once a token of the "leading" zone, i.e., the zone with
|
||||
// id 0 is determined, we must have enough space to accommodate the corresponding tokens in the
|
||||
// remaining (maxZonesCount-1) zones. Hence, the highest token of the leading zone must be a
|
||||
// multiple of maxZonesCount, that guarantees that there are remaining (maxZonesCount-1) available
|
||||
// tokens in the token space.
|
||||
maxTokenValue := uint32(((totalTokensCount / maxZonesCount) - 1) * maxZonesCount) |
||||
offset := maxTokenValue - token.prevToken |
||||
if offset < optimalTokenOwnership { |
||||
newToken := optimalTokenOwnership - offset |
||||
return newToken, nil |
||||
} |
||||
return token.prevToken + optimalTokenOwnership, nil |
||||
} |
||||
|
||||
// optimalTokenOwnership calculates the optimal ownership of the remaining currTokensCount tokens of an instance
|
||||
// having the given current instances ownership currInstanceOwnership and the given optimal instance ownership
|
||||
// optimalInstanceOwnership. The resulting token ownership must be a multiple of the number of zones.
|
||||
func (t *SpreadMinimizingTokenGenerator) optimalTokenOwnership(optimalInstanceOwnership, currInstanceOwnership float64, remainingTokensCount uint32) uint32 { |
||||
optimalTokenOwnership := uint32(optimalInstanceOwnership-currInstanceOwnership) / remainingTokensCount |
||||
return (optimalTokenOwnership / maxZonesCount) * maxZonesCount |
||||
} |
||||
|
||||
// GenerateTokens returns at most requestedTokensCount unique tokens, none of which clashes with the given
|
||||
// allTakenTokens, representing the set of all tokens currently present in the ring. Returned tokens are sorted.
|
||||
// The optimal number of tokens (optimalTokenPerInstance), i.e., 512, reserved for the underlying instance are
|
||||
// generated by generateAllTokens. GenerateTokens selects the first requestedTokensCount tokens from the reserved
|
||||
// tokens set, that are not already present in the takenTokens.
|
||||
// The number of returned tokens might be lower than the requested number of tokens in the following cases:
|
||||
// - if tokensCount is higher than 512 (optimalTokensPerInstance), or
|
||||
// - if among the 512 (optimalTokenPerInstance) reserved tokens there is less than tokenCount
|
||||
// tokens not already present in takenTokens.
|
||||
func (t *SpreadMinimizingTokenGenerator) GenerateTokens(requestedTokensCount int, allTakenTokens []uint32) Tokens { |
||||
used := make(map[uint32]bool, len(allTakenTokens)) |
||||
for _, v := range allTakenTokens { |
||||
used[v] = true |
||||
} |
||||
|
||||
allTokens := t.generateAllTokens() |
||||
uniqueTokens := make(Tokens, 0, requestedTokensCount) |
||||
|
||||
// allTokens is a sorted slice of tokens for instance t.cfg.InstanceID in zone t.cfg.zone
|
||||
// We filter out tokens from allTakenTokens, if any, and return at most requestedTokensCount tokens.
|
||||
for i := 0; i < len(allTokens) && len(uniqueTokens) < requestedTokensCount; i++ { |
||||
token := allTokens[i] |
||||
if used[token] { |
||||
continue |
||||
} |
||||
uniqueTokens = append(uniqueTokens, token) |
||||
} |
||||
return uniqueTokens |
||||
} |
||||
|
||||
// generateAllTokens generates the optimal number of tokens (optimalTokenPerInstance), i.e., 512,
|
||||
// for the underlying instance (with id t.instanceID). Generated tokens are sorted, and they are
|
||||
// distributed in such a way that registered ownership of the instance t.instanceID, when it is
|
||||
// placed in the ring that already contains instances with all the ids lower that t.instanceID
|
||||
// is optimal.
|
||||
// Calls to this method will always return the same set of tokens.
|
||||
func (t *SpreadMinimizingTokenGenerator) generateAllTokens() Tokens { |
||||
tokensByInstanceID := t.generateTokensByInstanceID() |
||||
allTokens := tokensByInstanceID[t.instanceID] |
||||
slices.Sort(allTokens) |
||||
return allTokens |
||||
} |
||||
|
||||
// generateTokensByInstanceID generates the optimal number of tokens (optimalTokenPerInstance),
|
||||
// i.e., 512, for all instances whose id is less or equal to the id of the underlying instance
|
||||
// (with id t.instanceID). Generated tokens are not sorted, but they are distributed in such a
|
||||
// way that registered ownership of all the instances is optimal.
|
||||
// Calls to this method will always return the same set of tokens.
|
||||
func (t *SpreadMinimizingTokenGenerator) generateTokensByInstanceID() map[int]Tokens { |
||||
firstInstanceTokens := t.generateFirstInstanceTokens() |
||||
tokensByInstanceID := make(map[int]Tokens, t.instanceID+1) |
||||
tokensByInstanceID[0] = firstInstanceTokens |
||||
|
||||
if t.instanceID == 0 { |
||||
return tokensByInstanceID |
||||
} |
||||
|
||||
// tokensQueues is a slice of priority queues. Slice indexes correspond
|
||||
// to the ids of instances, while priority queues represent the tokens
|
||||
// of the corresponding instance, ordered from highest to lowest ownership.
|
||||
tokensQueues := make([]ownershipPriorityQueue[ringToken], t.instanceID) |
||||
|
||||
// Create and initialize priority queue of tokens for the first instance
|
||||
tokensQueue := newPriorityQueue[ringToken](optimalTokensPerInstance) |
||||
prev := len(firstInstanceTokens) - 1 |
||||
firstInstanceOwnership := 0.0 |
||||
for tk, token := range firstInstanceTokens { |
||||
tokenOwnership := float64(tokenDistance(firstInstanceTokens[prev], token)) |
||||
firstInstanceOwnership += tokenOwnership |
||||
heap.Push(&tokensQueue, newRingTokenOwnershipInfo(token, firstInstanceTokens[prev])) |
||||
prev = tk |
||||
} |
||||
tokensQueues[0] = tokensQueue |
||||
|
||||
// instanceQueue is a priority queue of instances such that instances with higher ownership have a higher priority
|
||||
instanceQueue := newPriorityQueue[ringInstance](t.instanceID) |
||||
heap.Push(&instanceQueue, newRingInstanceOwnershipInfo(0, firstInstanceOwnership)) |
||||
|
||||
// ignoredInstances is a slice of the current instances whose tokens
|
||||
// don't have enough space to accommodate new tokens.
|
||||
ignoredInstances := make([]ownershipInfo[ringInstance], 0, t.instanceID) |
||||
|
||||
for i := 1; i <= t.instanceID; i++ { |
||||
optimalInstanceOwnership := float64(totalTokensCount) / float64(i+1) |
||||
currInstanceOwnership := 0.0 |
||||
addedTokens := 0 |
||||
ignoredInstances = ignoredInstances[:0] |
||||
tokens := make(Tokens, 0, optimalTokensPerInstance) |
||||
// currInstanceTokenQueue is the priority queue of tokens of newInstance
|
||||
currInstanceTokenQueue := newPriorityQueue[ringToken](optimalTokensPerInstance) |
||||
for addedTokens < optimalTokensPerInstance { |
||||
optimalTokenOwnership := t.optimalTokenOwnership(optimalInstanceOwnership, currInstanceOwnership, uint32(optimalTokensPerInstance-addedTokens)) |
||||
highestOwnershipInstance := instanceQueue.Peek() |
||||
if highestOwnershipInstance == nil || highestOwnershipInstance.ownership <= float64(optimalTokenOwnership) { |
||||
level.Warn(t.logger).Log("msg", "it was impossible to add a token because the instance with the highest ownership cannot satisfy the request", "added tokens", addedTokens+1, "highest ownership", highestOwnershipInstance.ownership, "requested ownership", optimalTokenOwnership) |
||||
// if this happens, it means that we cannot accommodate other tokens, so we panic
|
||||
err := fmt.Errorf("it was impossible to add %dth token for instance with id %d in zone %s because the instance with the highest ownership cannot satisfy the requested ownership %d", addedTokens+1, i, t.spreadMinimizingZones[t.zoneID], optimalTokenOwnership) |
||||
panic(err) |
||||
} |
||||
tokensQueue := tokensQueues[highestOwnershipInstance.item.instanceID] |
||||
highestOwnershipToken := tokensQueue.Peek() |
||||
if highestOwnershipToken.ownership <= float64(optimalTokenOwnership) { |
||||
// The token with the highest ownership of the instance with the highest ownership could not
|
||||
// accommodate a new token, hence we ignore this instance and pass to the next instance.
|
||||
ignoredInstances = append(ignoredInstances, heap.Pop(&instanceQueue).(ownershipInfo[ringInstance])) |
||||
continue |
||||
} |
||||
token := highestOwnershipToken.item |
||||
newToken, err := t.calculateNewToken(token, optimalTokenOwnership) |
||||
if err != nil { |
||||
level.Error(t.logger).Log("msg", "it was impossible to calculate a new token because an error occurred", "err", err) |
||||
// if this happens, it means that we cannot accommodate additional tokens, so we panic
|
||||
err := fmt.Errorf("it was impossible to calculate the %dth token for instance with id %d in zone %s", addedTokens+1, i, t.spreadMinimizingZones[t.zoneID]) |
||||
panic(err) |
||||
} |
||||
tokens = append(tokens, newToken) |
||||
// add the new token to currInstanceTokenQueue
|
||||
heap.Push(&currInstanceTokenQueue, newRingTokenOwnershipInfo(newToken, token.prevToken)) |
||||
|
||||
oldTokenOwnership := highestOwnershipToken.ownership |
||||
newTokenOwnership := float64(tokenDistance(newToken, token.token)) |
||||
currInstanceOwnership += oldTokenOwnership - newTokenOwnership |
||||
|
||||
// The token with the highest ownership of the instance with the highest ownership has changed,
|
||||
// so we propagate these changes in the corresponding tokens queue.
|
||||
highestOwnershipToken.item.prevToken = newToken |
||||
highestOwnershipToken.ownership = newTokenOwnership |
||||
heap.Fix(&tokensQueue, 0) |
||||
|
||||
// The ownership of the instance with the highest ownership has changed,
|
||||
// so we propagate these changes in the instances queue.
|
||||
highestOwnershipInstance.ownership = highestOwnershipInstance.ownership - oldTokenOwnership + newTokenOwnership |
||||
heap.Fix(&instanceQueue, 0) |
||||
|
||||
addedTokens++ |
||||
} |
||||
tokensByInstanceID[i] = tokens |
||||
// if this is the last iteration we return, so we avoid to call additional heap.Pushs
|
||||
if i == t.instanceID { |
||||
return tokensByInstanceID |
||||
} |
||||
|
||||
// If there were some ignored instances, we put them back on the queue.
|
||||
for _, ignoredInstance := range ignoredInstances { |
||||
heap.Push(&instanceQueue, ignoredInstance) |
||||
} |
||||
|
||||
tokensQueues[i] = currInstanceTokenQueue |
||||
|
||||
// add the current instance with the calculated ownership currInstanceOwnership to instanceQueue
|
||||
heap.Push(&instanceQueue, newRingInstanceOwnershipInfo(i, currInstanceOwnership)) |
||||
} |
||||
|
||||
return tokensByInstanceID |
||||
} |
||||
@ -0,0 +1,54 @@ |
||||
package ring |
||||
|
||||
import ( |
||||
"math/rand" |
||||
"sort" |
||||
"time" |
||||
) |
||||
|
||||
type TokenGenerator interface { |
||||
// GenerateTokens generates at most requestedTokensCount unique tokens, none of which clashes with
|
||||
// the given allTakenTokens, representing the set of all tokens currently present in the ring.
|
||||
// Generated tokens are sorted.
|
||||
GenerateTokens(requestedTokensCount int, allTakenTokens []uint32) Tokens |
||||
} |
||||
|
||||
type RandomTokenGenerator struct{} |
||||
|
||||
func NewRandomTokenGenerator() *RandomTokenGenerator { |
||||
return &RandomTokenGenerator{} |
||||
} |
||||
|
||||
// GenerateTokens generates at most requestedTokensCount unique random tokens, none of which clashes with
|
||||
// the given allTakenTokens, representing the set of all tokens currently present in the ring.
|
||||
// Generated tokens are sorted.
|
||||
func (t *RandomTokenGenerator) GenerateTokens(requestedTokensCount int, allTakenTokens []uint32) Tokens { |
||||
if requestedTokensCount <= 0 { |
||||
return []uint32{} |
||||
} |
||||
|
||||
r := rand.New(rand.NewSource(time.Now().UnixNano())) |
||||
|
||||
used := make(map[uint32]bool, len(allTakenTokens)) |
||||
for _, v := range allTakenTokens { |
||||
used[v] = true |
||||
} |
||||
|
||||
tokens := make([]uint32, 0, requestedTokensCount) |
||||
for i := 0; i < requestedTokensCount; { |
||||
candidate := r.Uint32() |
||||
if used[candidate] { |
||||
continue |
||||
} |
||||
used[candidate] = true |
||||
tokens = append(tokens, candidate) |
||||
i++ |
||||
} |
||||
|
||||
// Ensure returned tokens are sorted.
|
||||
sort.Slice(tokens, func(i, j int) bool { |
||||
return tokens[i] < tokens[j] |
||||
}) |
||||
|
||||
return tokens |
||||
} |
||||
Loading…
Reference in new issue