satellite/orders: add settled amount to rollups write cache
https://storjlabs.atlassian.net/browse/SM-1109 Change-Id: Ic5859b141c1384157b33df0d7fb6c8b43cc8a6b1
This commit is contained in:
parent
12a15e5a6a
commit
41497569cd
@ -20,6 +20,7 @@ import (
|
||||
type CacheData struct {
|
||||
Inline int64
|
||||
Allocated int64
|
||||
Settled int64
|
||||
}
|
||||
|
||||
// CacheKey is the key information for the cached map below
|
||||
@ -61,12 +62,17 @@ func NewRollupsWriteCache(log *zap.Logger, db DB, batchSize int) *RollupsWriteCa
|
||||
|
||||
// UpdateBucketBandwidthAllocation updates the rollups cache adding allocated data for a bucket bandwidth rollup
|
||||
func (cache *RollupsWriteCache) UpdateBucketBandwidthAllocation(ctx context.Context, projectID uuid.UUID, bucketName []byte, action pb.PieceAction, amount int64, intervalStart time.Time) error {
|
||||
return cache.updateCacheValue(ctx, projectID, bucketName, action, amount, 0, intervalStart.UTC())
|
||||
return cache.updateCacheValue(ctx, projectID, bucketName, action, amount, 0, 0, intervalStart.UTC())
|
||||
}
|
||||
|
||||
// UpdateBucketBandwidthInline updates the rollups cache adding inline data for a bucket bandwidth rollup
|
||||
func (cache *RollupsWriteCache) UpdateBucketBandwidthInline(ctx context.Context, projectID uuid.UUID, bucketName []byte, action pb.PieceAction, amount int64, intervalStart time.Time) error {
|
||||
return cache.updateCacheValue(ctx, projectID, bucketName, action, 0, amount, intervalStart.UTC())
|
||||
return cache.updateCacheValue(ctx, projectID, bucketName, action, 0, amount, 0, intervalStart.UTC())
|
||||
}
|
||||
|
||||
// UpdateBucketBandwidthSettle updates the rollups cache adding settled data for a bucket bandwidth rollup
|
||||
func (cache *RollupsWriteCache) UpdateBucketBandwidthSettle(ctx context.Context, projectID uuid.UUID, bucketName []byte, action pb.PieceAction, amount int64, intervalStart time.Time) error {
|
||||
return cache.updateCacheValue(ctx, projectID, bucketName, action, 0, 0, amount, intervalStart.UTC())
|
||||
}
|
||||
|
||||
// resetCache should only be called after you have acquired the cache lock. It
|
||||
@ -114,6 +120,7 @@ func (cache *RollupsWriteCache) flush(ctx context.Context, pendingRollups Rollup
|
||||
Action: cacheKey.Action,
|
||||
Inline: cacheData.Inline,
|
||||
Allocated: cacheData.Allocated,
|
||||
Settled: cacheData.Settled,
|
||||
})
|
||||
}
|
||||
|
||||
@ -131,7 +138,7 @@ func (cache *RollupsWriteCache) flush(ctx context.Context, pendingRollups Rollup
|
||||
completion.Release()
|
||||
}
|
||||
|
||||
func (cache *RollupsWriteCache) updateCacheValue(ctx context.Context, projectID uuid.UUID, bucketName []byte, action pb.PieceAction, allocated, inline int64, intervalStart time.Time) error {
|
||||
func (cache *RollupsWriteCache) updateCacheValue(ctx context.Context, projectID uuid.UUID, bucketName []byte, action pb.PieceAction, allocated, inline, settled int64, intervalStart time.Time) error {
|
||||
defer mon.Task()(&ctx)(nil)
|
||||
|
||||
cache.mu.Lock()
|
||||
@ -157,6 +164,7 @@ func (cache *RollupsWriteCache) updateCacheValue(ctx context.Context, projectID
|
||||
}
|
||||
data.Allocated += allocated
|
||||
data.Inline += inline
|
||||
data.Settled += settled
|
||||
cache.pendingRollups[key] = data
|
||||
|
||||
if cache.currentSize < cache.batchSize {
|
||||
|
@ -129,7 +129,7 @@ func TestRollupsWriteCacheBatchChore(t *testing.T) {
|
||||
)
|
||||
}
|
||||
|
||||
func TestUpdateBucketBandwidthAllocation(t *testing.T) {
|
||||
func TestUpdateBucketBandwidth(t *testing.T) {
|
||||
testplanet.Run(t, testplanet.Config{
|
||||
SatelliteCount: 1,
|
||||
},
|
||||
@ -142,52 +142,68 @@ func TestUpdateBucketBandwidthAllocation(t *testing.T) {
|
||||
cache, ok := ordersDB.(*orders.RollupsWriteCache)
|
||||
require.True(t, ok)
|
||||
size := cache.CurrentSize()
|
||||
require.Equal(t, size, 0)
|
||||
require.Equal(t, 0, size)
|
||||
|
||||
// setup: add one item to the cache
|
||||
// setup: add an allocated and settled item to the cache
|
||||
projectID := testrand.UUID()
|
||||
bucketName := []byte("testbucketname")
|
||||
amount := (memory.MB * 500).Int64()
|
||||
err := ordersDB.UpdateBucketBandwidthAllocation(ctx, projectID, bucketName, pb.PieceAction_GET, amount, time.Now())
|
||||
require.NoError(t, err)
|
||||
err = ordersDB.UpdateBucketBandwidthSettle(ctx, projectID, bucketName, pb.PieceAction_PUT, amount, time.Now())
|
||||
require.NoError(t, err)
|
||||
|
||||
// test: confirm there is one item in the cache now
|
||||
size = cache.CurrentSize()
|
||||
require.Equal(t, size, 1)
|
||||
require.Equal(t, 2, size)
|
||||
projectMap := cache.CurrentData()
|
||||
expectedKey := orders.CacheKey{
|
||||
expectedKeyAllocated := orders.CacheKey{
|
||||
ProjectID: projectID,
|
||||
BucketName: string(bucketName),
|
||||
Action: pb.PieceAction_GET,
|
||||
}
|
||||
expected := orders.RollupData{
|
||||
expectedKey: {
|
||||
Inline: 0,
|
||||
Allocated: amount,
|
||||
},
|
||||
expectedKeySettled := orders.CacheKey{
|
||||
ProjectID: projectID,
|
||||
BucketName: string(bucketName),
|
||||
Action: pb.PieceAction_PUT,
|
||||
}
|
||||
require.Equal(t, projectMap, expected)
|
||||
expectedCacheDataAllocated := orders.CacheData{
|
||||
Inline: 0,
|
||||
Allocated: amount,
|
||||
Settled: 0,
|
||||
}
|
||||
expectedCacheDataSettled := orders.CacheData{
|
||||
Inline: 0,
|
||||
Allocated: 0,
|
||||
Settled: amount,
|
||||
}
|
||||
require.Equal(t, projectMap[expectedKeyAllocated], expectedCacheDataAllocated)
|
||||
require.Equal(t, projectMap[expectedKeySettled], expectedCacheDataSettled)
|
||||
|
||||
// setup: add another item to the cache but with a different projectID
|
||||
projectID2 := testrand.UUID()
|
||||
amount2 := (memory.MB * 10).Int64()
|
||||
err = ordersDB.UpdateBucketBandwidthAllocation(ctx, projectID2, bucketName, pb.PieceAction_GET, amount2, time.Now())
|
||||
require.NoError(t, err)
|
||||
err = ordersDB.UpdateBucketBandwidthSettle(ctx, projectID2, bucketName, pb.PieceAction_GET, amount2, time.Now())
|
||||
require.NoError(t, err)
|
||||
size = cache.CurrentSize()
|
||||
require.Equal(t, size, 2)
|
||||
require.Equal(t, 3, size)
|
||||
projectMap2 := cache.CurrentData()
|
||||
// test: confirm there are two items in the cache now for different projectIDs
|
||||
|
||||
expectedKey = orders.CacheKey{
|
||||
// test: confirm there are 3 items in the cache now with different projectIDs
|
||||
expectedKey := orders.CacheKey{
|
||||
ProjectID: projectID2,
|
||||
BucketName: string(bucketName),
|
||||
Action: pb.PieceAction_GET,
|
||||
}
|
||||
expected[expectedKey] = orders.CacheData{
|
||||
expectedData := orders.CacheData{
|
||||
Inline: 0,
|
||||
Allocated: amount2,
|
||||
Settled: amount2,
|
||||
}
|
||||
require.Equal(t, projectMap2, expected)
|
||||
require.Equal(t, projectMap2[expectedKey], expectedData)
|
||||
require.Equal(t, len(projectMap2), 3)
|
||||
},
|
||||
)
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user