mirror of https://github.com/grafana/loki
Added max streams per user global limit (#1493)
* Added max streams per user global limit Signed-off-by: Marco Pracucci <marco@pracucci.com> * Updated changelog Signed-off-by: Marco Pracucci <marco@pracucci.com> Co-authored-by: Cyril Tovena <cyril.tovena@gmail.com>pull/1499/head
Marco Pracucci
6 years ago
committed by
Cyril Tovena
parent
ce407d3276
commit
ec40515d31
@ -0,0 +1,94 @@ |
||||
package ingester |
||||
|
||||
import ( |
||||
"fmt" |
||||
"math" |
||||
|
||||
"github.com/grafana/loki/pkg/util/validation" |
||||
) |
||||
|
||||
const ( |
||||
errMaxStreamsPerUserLimitExceeded = "per-user streams limit (local: %d global: %d actual local: %d) exceeded" |
||||
) |
||||
|
||||
// RingCount is the interface exposed by a ring implementation which allows
|
||||
// to count members
|
||||
type RingCount interface { |
||||
HealthyInstancesCount() int |
||||
} |
||||
|
||||
// Limiter implements primitives to get the maximum number of streams
|
||||
// an ingester can handle for a specific tenant
|
||||
type Limiter struct { |
||||
limits *validation.Overrides |
||||
ring RingCount |
||||
replicationFactor int |
||||
} |
||||
|
||||
// NewLimiter makes a new limiter
|
||||
func NewLimiter(limits *validation.Overrides, ring RingCount, replicationFactor int) *Limiter { |
||||
return &Limiter{ |
||||
limits: limits, |
||||
ring: ring, |
||||
replicationFactor: replicationFactor, |
||||
} |
||||
} |
||||
|
||||
// AssertMaxStreamsPerUser ensures limit has not been reached compared to the current
|
||||
// number of streams in input and returns an error if so.
|
||||
func (l *Limiter) AssertMaxStreamsPerUser(userID string, streams int) error { |
||||
actualLimit := l.maxStreamsPerUser(userID) |
||||
if streams < actualLimit { |
||||
return nil |
||||
} |
||||
|
||||
localLimit := l.limits.MaxLocalStreamsPerUser(userID) |
||||
globalLimit := l.limits.MaxGlobalStreamsPerUser(userID) |
||||
|
||||
return fmt.Errorf(errMaxStreamsPerUserLimitExceeded, localLimit, globalLimit, actualLimit) |
||||
} |
||||
|
||||
func (l *Limiter) maxStreamsPerUser(userID string) int { |
||||
localLimit := l.limits.MaxLocalStreamsPerUser(userID) |
||||
|
||||
// We can assume that streams are evenly distributed across ingesters
|
||||
// so we do convert the global limit into a local limit
|
||||
globalLimit := l.limits.MaxGlobalStreamsPerUser(userID) |
||||
localLimit = l.minNonZero(localLimit, l.convertGlobalToLocalLimit(globalLimit)) |
||||
|
||||
// If both the local and global limits are disabled, we just
|
||||
// use the largest int value
|
||||
if localLimit == 0 { |
||||
localLimit = math.MaxInt32 |
||||
} |
||||
|
||||
return localLimit |
||||
} |
||||
|
||||
func (l *Limiter) convertGlobalToLocalLimit(globalLimit int) int { |
||||
if globalLimit == 0 { |
||||
return 0 |
||||
} |
||||
|
||||
// Given we don't need a super accurate count (ie. when the ingesters
|
||||
// topology changes) and we prefer to always be in favor of the tenant,
|
||||
// we can use a per-ingester limit equal to:
|
||||
// (global limit / number of ingesters) * replication factor
|
||||
numIngesters := l.ring.HealthyInstancesCount() |
||||
|
||||
// May happen because the number of ingesters is asynchronously updated.
|
||||
// If happens, we just temporarily ignore the global limit.
|
||||
if numIngesters > 0 { |
||||
return int((float64(globalLimit) / float64(numIngesters)) * float64(l.replicationFactor)) |
||||
} |
||||
|
||||
return 0 |
||||
} |
||||
|
||||
func (l *Limiter) minNonZero(first, second int) int { |
||||
if first == 0 || (second != 0 && first > second) { |
||||
return second |
||||
} |
||||
|
||||
return first |
||||
} |
||||
@ -0,0 +1,195 @@ |
||||
package ingester |
||||
|
||||
import ( |
||||
"fmt" |
||||
"math" |
||||
"testing" |
||||
|
||||
"github.com/grafana/loki/pkg/util/validation" |
||||
"github.com/stretchr/testify/assert" |
||||
"github.com/stretchr/testify/require" |
||||
) |
||||
|
||||
func TestLimiter_maxStreamsPerUser(t *testing.T) { |
||||
tests := map[string]struct { |
||||
maxLocalStreamsPerUser int |
||||
maxGlobalStreamsPerUser int |
||||
ringReplicationFactor int |
||||
ringIngesterCount int |
||||
expected int |
||||
}{ |
||||
"both local and global limits are disabled": { |
||||
maxLocalStreamsPerUser: 0, |
||||
maxGlobalStreamsPerUser: 0, |
||||
ringReplicationFactor: 1, |
||||
ringIngesterCount: 1, |
||||
expected: math.MaxInt32, |
||||
}, |
||||
"only local limit is enabled": { |
||||
maxLocalStreamsPerUser: 1000, |
||||
maxGlobalStreamsPerUser: 0, |
||||
ringReplicationFactor: 1, |
||||
ringIngesterCount: 1, |
||||
expected: 1000, |
||||
}, |
||||
"only global limit is enabled with replication-factor=1": { |
||||
maxLocalStreamsPerUser: 0, |
||||
maxGlobalStreamsPerUser: 1000, |
||||
ringReplicationFactor: 1, |
||||
ringIngesterCount: 10, |
||||
expected: 100, |
||||
}, |
||||
"only global limit is enabled with replication-factor=3": { |
||||
maxLocalStreamsPerUser: 0, |
||||
maxGlobalStreamsPerUser: 1000, |
||||
ringReplicationFactor: 3, |
||||
ringIngesterCount: 10, |
||||
expected: 300, |
||||
}, |
||||
"both local and global limits are set with local limit < global limit": { |
||||
maxLocalStreamsPerUser: 150, |
||||
maxGlobalStreamsPerUser: 1000, |
||||
ringReplicationFactor: 3, |
||||
ringIngesterCount: 10, |
||||
expected: 150, |
||||
}, |
||||
"both local and global limits are set with local limit > global limit": { |
||||
maxLocalStreamsPerUser: 500, |
||||
maxGlobalStreamsPerUser: 1000, |
||||
ringReplicationFactor: 3, |
||||
ringIngesterCount: 10, |
||||
expected: 300, |
||||
}, |
||||
} |
||||
|
||||
for testName, testData := range tests { |
||||
testData := testData |
||||
|
||||
t.Run(testName, func(t *testing.T) { |
||||
// Mock the ring
|
||||
ring := &ringCountMock{count: testData.ringIngesterCount} |
||||
|
||||
// Mock limits
|
||||
limits, err := validation.NewOverrides(validation.Limits{ |
||||
MaxLocalStreamsPerUser: testData.maxLocalStreamsPerUser, |
||||
MaxGlobalStreamsPerUser: testData.maxGlobalStreamsPerUser, |
||||
}) |
||||
require.NoError(t, err) |
||||
|
||||
limiter := NewLimiter(limits, ring, testData.ringReplicationFactor) |
||||
actual := limiter.maxStreamsPerUser("test") |
||||
|
||||
assert.Equal(t, testData.expected, actual) |
||||
}) |
||||
} |
||||
} |
||||
|
||||
func TestLimiter_AssertMaxStreamsPerUser(t *testing.T) { |
||||
tests := map[string]struct { |
||||
maxLocalStreamsPerUser int |
||||
maxGlobalStreamsPerUser int |
||||
ringReplicationFactor int |
||||
ringIngesterCount int |
||||
streams int |
||||
expected error |
||||
}{ |
||||
"both local and global limit are disabled": { |
||||
maxLocalStreamsPerUser: 0, |
||||
maxGlobalStreamsPerUser: 0, |
||||
ringReplicationFactor: 1, |
||||
ringIngesterCount: 1, |
||||
streams: 100, |
||||
expected: nil, |
||||
}, |
||||
"current number of streams is below the limit": { |
||||
maxLocalStreamsPerUser: 0, |
||||
maxGlobalStreamsPerUser: 1000, |
||||
ringReplicationFactor: 3, |
||||
ringIngesterCount: 10, |
||||
streams: 299, |
||||
expected: nil, |
||||
}, |
||||
"current number of streams is above the limit": { |
||||
maxLocalStreamsPerUser: 0, |
||||
maxGlobalStreamsPerUser: 1000, |
||||
ringReplicationFactor: 3, |
||||
ringIngesterCount: 10, |
||||
streams: 300, |
||||
expected: fmt.Errorf(errMaxStreamsPerUserLimitExceeded, 0, 1000, 300), |
||||
}, |
||||
} |
||||
|
||||
for testName, testData := range tests { |
||||
testData := testData |
||||
|
||||
t.Run(testName, func(t *testing.T) { |
||||
// Mock the ring
|
||||
ring := &ringCountMock{count: testData.ringIngesterCount} |
||||
|
||||
// Mock limits
|
||||
limits, err := validation.NewOverrides(validation.Limits{ |
||||
MaxLocalStreamsPerUser: testData.maxLocalStreamsPerUser, |
||||
MaxGlobalStreamsPerUser: testData.maxGlobalStreamsPerUser, |
||||
}) |
||||
require.NoError(t, err) |
||||
|
||||
limiter := NewLimiter(limits, ring, testData.ringReplicationFactor) |
||||
actual := limiter.AssertMaxStreamsPerUser("test", testData.streams) |
||||
|
||||
assert.Equal(t, testData.expected, actual) |
||||
}) |
||||
} |
||||
} |
||||
|
||||
func TestLimiter_minNonZero(t *testing.T) { |
||||
t.Parallel() |
||||
|
||||
tests := map[string]struct { |
||||
first int |
||||
second int |
||||
expected int |
||||
}{ |
||||
"both zero": { |
||||
first: 0, |
||||
second: 0, |
||||
expected: 0, |
||||
}, |
||||
"first is zero": { |
||||
first: 0, |
||||
second: 1, |
||||
expected: 1, |
||||
}, |
||||
"second is zero": { |
||||
first: 1, |
||||
second: 0, |
||||
expected: 1, |
||||
}, |
||||
"both non zero, second > first": { |
||||
first: 1, |
||||
second: 2, |
||||
expected: 1, |
||||
}, |
||||
"both non zero, first > second": { |
||||
first: 2, |
||||
second: 1, |
||||
expected: 1, |
||||
}, |
||||
} |
||||
|
||||
for testName, testData := range tests { |
||||
testData := testData |
||||
|
||||
t.Run(testName, func(t *testing.T) { |
||||
limiter := NewLimiter(nil, nil, 0) |
||||
assert.Equal(t, testData.expected, limiter.minNonZero(testData.first, testData.second)) |
||||
}) |
||||
} |
||||
} |
||||
|
||||
type ringCountMock struct { |
||||
count int |
||||
} |
||||
|
||||
func (m *ringCountMock) HealthyInstancesCount() int { |
||||
return m.count |
||||
} |
||||
Loading…
Reference in new issue