Skip to content

Commit 234c526

Browse files
juliusvjuliusv
committed
Allow switching to daily instead of hourly index buckets
There can be slight overlap around the switchover time, but I deemed it safer (and not incorrect) than risking gaps due to some calculation error.
1 parent 5318a4e commit 234c526

File tree

3 files changed

+113
-41
lines changed

3 files changed

+113
-41
lines changed

‎chunk/chunk_store.go‎

Lines changed: 48 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66
"math/rand"
77
"net/url"
88
"sort"
9+
"strconv"
910
"strings"
1011
"sync"
1112
"sync/atomic"
@@ -139,9 +140,10 @@ type Store interface {
139140

140141
// StoreConfig specifies config for a ChunkStore
141142
type StoreConfig struct {
142-
S3URL string
143-
DynamoDBURL string
144-
ChunkCache *Cache
143+
S3URL string
144+
DynamoDBURL string
145+
ChunkCache *Cache
146+
DailyBucketsFrom model.Time
145147

146148
// Not exported as only used by tests to inject mocks
147149
dynamodb dynamodbClient
@@ -206,6 +208,10 @@ type AWSStore struct {
206208
tableName string
207209
bucketName string
208210

211+
// Around which time to start bucketing indexes by day instead of by hour.
212+
// Only the day matters, not the time within the day.
213+
dailyBucketsFrom model.Time
214+
209215
dynamoRequests chan dynamoOp
210216
dynamoRequestsDone sync.WaitGroup
211217
}
@@ -332,25 +338,45 @@ func (c *AWSStore) CreateTables() error {
332338
return err
333339
}
334340

335-
func bigBuckets(from, through model.Time) []int64 {
341+
func bigBuckets(from, through model.Time, dailyBucketsFrom model.Time) []string {
336342
var (
337343
secondsInHour = int64(time.Hour / time.Second)
338344
fromHour = from.Unix() / secondsInHour
339345
throughHour = through.Unix() / secondsInHour
340-
result []int64
346+
347+
secondsInDay = int64(24 * time.Hour / time.Second)
348+
fromDay = from.Unix() / secondsInDay
349+
throughDay = through.Unix() / secondsInDay
350+
351+
firstDailyBucket = dailyBucketsFrom.Unix() / secondsInDay
352+
lastHourlyBucket = firstDailyBucket * 24
353+
354+
result []string
341355
)
356+
342357
for i := fromHour; i <= throughHour; i++ {
343-
result = append(result, i)
358+
if i > lastHourlyBucket {
359+
break
360+
}
361+
result = append(result, strconv.Itoa(int(i)))
344362
}
363+
364+
for i := fromDay; i <= throughDay; i++ {
365+
if i < firstDailyBucket {
366+
continue
367+
}
368+
result = append(result, fmt.Sprintf("d%d", int(i)))
369+
}
370+
345371
return result
346372
}
347373

348374
func chunkName(userID, chunkID string) string {
349375
return fmt.Sprintf("%s/%s", userID, chunkID)
350376
}
351377

352-
func hashValue(userID string, hour int64, metricName model.LabelValue) string {
353-
return fmt.Sprintf("%s:%d:%s", userID, hour, metricName)
378+
func hashValue(userID string, bucket string, metricName model.LabelValue) string {
379+
return fmt.Sprintf("%s:%s:%s", userID, bucket, metricName)
354380
}
355381

356382
func rangeValue(label model.LabelName, value model.LabelValue, chunkID string) []byte {
@@ -456,8 +482,8 @@ func (c *AWSStore) calculateDynamoWrites(userID string, chunks []Chunk) ([]*dyna
456482
}
457483

458484
entries := 0
459-
for _, hour := range bigBuckets(chunk.From, chunk.Through) {
460-
hashValue := hashValue(userID, hour, metricName)
485+
for _, bucket := range bigBuckets(chunk.From, chunk.Through, c.dailyBucketsFrom) {
486+
hashValue := hashValue(userID, bucket, metricName)
461487
for label, value := range chunk.Metric {
462488
if label == model.MetricNameLabel {
463489
continue
@@ -556,18 +582,18 @@ func (c *AWSStore) lookupChunks(ctx context.Context, userID string, from, throug
556582

557583
incomingChunkSets := make(chan ByID)
558584
incomingErrors := make(chan error)
559-
buckets := bigBuckets(from, through)
585+
buckets := bigBuckets(from, through, c.dailyBucketsFrom)
560586
totalLookups := int32(0)
561-
for _, hour := range buckets {
562-
go func(hour int64) {
563-
incoming, lookups, err := c.lookupChunksFor(ctx, userID, hour, metricName, matchers)
587+
for _, b := range buckets {
588+
go func(bucket string) {
589+
incoming, lookups, err := c.lookupChunksFor(ctx, userID, bucket, metricName, matchers)
564590
atomic.AddInt32(&totalLookups, lookups)
565591
if err != nil {
566592
incomingErrors <- err
567593
} else {
568594
incomingChunkSets <- incoming
569595
}
570-
}(hour)
596+
}(b)
571597
}
572598

573599
var chunks ByID
@@ -592,17 +618,17 @@ func next(s string) string {
592618
return result
593619
}
594620

595-
func (c *AWSStore) lookupChunksFor(ctx context.Context, userID string, hour int64, metricName model.LabelValue, matchers []*metric.LabelMatcher) (ByID, int32, error) {
621+
func (c *AWSStore) lookupChunksFor(ctx context.Context, userID string, bucket string, metricName model.LabelValue, matchers []*metric.LabelMatcher) (ByID, int32, error) {
596622
if len(matchers) == 0 {
597-
return c.lookupChunksForMetricName(ctx, userID, hour, metricName)
623+
return c.lookupChunksForMetricName(ctx, userID, bucket, metricName)
598624
}
599625

600626
incomingChunkSets := make(chan ByID)
601627
incomingErrors := make(chan error)
602628

603629
for _, matcher := range matchers {
604630
go func(matcher *metric.LabelMatcher) {
605-
incoming, err := c.lookupChunksForMatcher(ctx, userID, hour, metricName, matcher)
631+
incoming, err := c.lookupChunksForMatcher(ctx, userID, bucket, metricName, matcher)
606632
if err != nil {
607633
incomingErrors <- err
608634
} else {
@@ -624,8 +650,8 @@ func (c *AWSStore) lookupChunksFor(ctx context.Context, userID string, hour int6
624650
return nWayIntersect(chunkSets), int32(len(matchers)), lastErr
625651
}
626652

627-
func (c *AWSStore) lookupChunksForMetricName(ctx context.Context, userID string, hour int64, metricName model.LabelValue) (ByID, int32, error) {
628-
hashValue := hashValue(userID, hour, metricName)
653+
func (c *AWSStore) lookupChunksForMetricName(ctx context.Context, userID string, bucket string, metricName model.LabelValue) (ByID, int32, error) {
654+
hashValue := hashValue(userID, bucket, metricName)
629655
input := &dynamodb.QueryInput{
630656
TableName: aws.String(c.tableName),
631657
KeyConditions: map[string]*dynamodb.Condition{
@@ -665,8 +691,8 @@ func (c *AWSStore) lookupChunksForMetricName(ctx context.Context, userID string,
665691
return chunkSet, 1, nil
666692
}
667693

668-
func (c *AWSStore) lookupChunksForMatcher(ctx context.Context, userID string, hour int64, metricName model.LabelValue, matcher *metric.LabelMatcher) (ByID, error) {
669-
hashValue := hashValue(userID, hour, metricName)
694+
func (c *AWSStore) lookupChunksForMatcher(ctx context.Context, userID string, bucket string, metricName model.LabelValue, matcher *metric.LabelMatcher) (ByID, error) {
695+
hashValue := hashValue(userID, bucket, metricName)
670696
var rangeMinValue, rangeMaxValue []byte
671697
if matcher.Type == metric.Equal {
672698
nextValue := model.LabelValue(next(string(matcher.Value)))

‎chunk/chunk_store_test.go‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -146,3 +146,41 @@ func diff(want, have interface{}) string {
146146
})
147147
return "\n" + text
148148
}
149+
150+
func TestBigBuckets(t *testing.T) {
151+
scenarios := []struct {
152+
from, through, dayEpochFrom model.Time
153+
buckets []string
154+
}{
155+
{
156+
from: model.TimeFromUnix(0),
157+
through: model.TimeFromUnix(0).Add(3*24*time.Hour) - 1,
158+
dayEpochFrom: model.TimeFromUnix(0).Add(1 * 24 * time.Hour),
159+
buckets: []string{"0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18", "19", "20", "21", "22", "23", "24", "d1", "d2"},
160+
},
161+
{
162+
from: model.TimeFromUnix(0),
163+
through: model.TimeFromUnix(0).Add(3*24*time.Hour) - 1,
164+
dayEpochFrom: model.TimeFromUnix(0).Add(2*24*time.Hour) - 1, // Only the day matters for the epoch start time, not the time.
165+
buckets: []string{"0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18", "19", "20", "21", "22", "23", "24", "d1", "d2"},
166+
},
167+
{
168+
from: model.TimeFromUnix(0),
169+
through: model.TimeFromUnix(0).Add(3*24*time.Hour) - 1,
170+
dayEpochFrom: model.TimeFromUnix(0).Add(1*24*time.Hour) - 1,
171+
buckets: []string{"0", "d0", "d1", "d2"},
172+
},
173+
{
174+
from: model.TimeFromUnix(0),
175+
through: model.TimeFromUnix(0).Add(2 * 24 * time.Hour),
176+
dayEpochFrom: model.TimeFromUnix(0).Add(99 * 24 * time.Hour),
177+
buckets: []string{"0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18", "19", "20", "21", "22", "23", "24", "25", "26", "27", "28", "29", "30", "31", "32", "33", "34", "35", "36", "37", "38", "39", "40", "41", "42", "43", "44", "45", "46", "47", "48"},
178+
},
179+
}
180+
for i, s := range scenarios {
181+
buckets := bigBuckets(s.from, s.through, s.dayEpochFrom)
182+
if !reflect.DeepEqual(buckets, s.buckets) {
183+
t.Fatalf("%d. unexpected buckets; want %v, got %v", i, s.buckets, buckets)
184+
}
185+
}
186+
}

‎cmd/cortex/main.go‎

Lines changed: 27 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import (
2222

2323
"github.com/prometheus/client_golang/prometheus"
2424
"github.com/prometheus/common/log"
25+
"github.com/prometheus/common/model"
2526
"github.com/prometheus/common/route"
2627
"github.com/prometheus/prometheus/promql"
2728
"github.com/prometheus/prometheus/web/api/v1"
@@ -60,22 +61,23 @@ func init() {
6061
}
6162

6263
type cfg struct {
63-
mode string
64-
listenPort int
65-
consulHost string
66-
consulPrefix string
67-
s3URL string
68-
dynamodbURL string
69-
dynamodbCreateTables bool
70-
dynamodbPollInterval time.Duration
71-
memcachedHostname string
72-
memcachedTimeout time.Duration
73-
memcachedExpiration time.Duration
74-
memcachedService string
75-
remoteTimeout time.Duration
76-
numTokens int
77-
logSuccess bool
78-
watchDynamo bool
64+
mode string
65+
listenPort int
66+
consulHost string
67+
consulPrefix string
68+
s3URL string
69+
dynamodbURL string
70+
dynamodbCreateTables bool
71+
dynamodbPollInterval time.Duration
72+
dynamodbDailyBucketsFrom string
73+
memcachedHostname string
74+
memcachedTimeout time.Duration
75+
memcachedExpiration time.Duration
76+
memcachedService string
77+
remoteTimeout time.Duration
78+
numTokens int
79+
logSuccess bool
80+
watchDynamo bool
7981

8082
ingesterConfig ingester.Config
8183
distributorConfig distributor.Config
@@ -95,6 +97,7 @@ func main() {
9597
flag.StringVar(&cfg.dynamodbURL, "dynamodb.url", "localhost:8000", "DynamoDB endpoint URL.")
9698
flag.DurationVar(&cfg.dynamodbPollInterval, "dynamodb.poll-interval", 2*time.Minute, "How frequently to poll DynamoDB to learn our capacity.")
9799
flag.BoolVar(&cfg.dynamodbCreateTables, "dynamodb.create-tables", false, "Create required DynamoDB tables on startup.")
100+
flag.StringVar(&cfg.dynamodbDailyBucketsFrom, "dynamodb.daily-buckets-from", "9999-01-01", "The date in the format YYYY-MM-DD of the first day for which DynamoDB index buckets should be day-sized vs. hour-sized.")
98101
flag.BoolVar(&cfg.watchDynamo, "watch-dynamo", false, "Periodically collect DynamoDB provisioned throughput.")
99102

100103
flag.StringVar(&cfg.memcachedHostname, "memcached.hostname", "", "Hostname for memcached service to use when caching chunks. If empty, no memcached will be used.")
@@ -246,10 +249,15 @@ func setupChunkStore(cfg cfg) (chunk.Store, error) {
246249
Expiration: cfg.memcachedExpiration,
247250
}
248251
}
252+
dailyBucketsFrom, err := time.Parse("2006-01-02", cfg.dynamodbDailyBucketsFrom)
253+
if err != nil {
254+
return nil, fmt.Errorf("error parsing daily buckets begin date: %v", err)
255+
}
249256
chunkStore, err := chunk.NewAWSStore(chunk.StoreConfig{
250-
S3URL: cfg.s3URL,
251-
DynamoDBURL: cfg.dynamodbURL,
252-
ChunkCache: chunkCache,
257+
S3URL: cfg.s3URL,
258+
DynamoDBURL: cfg.dynamodbURL,
259+
ChunkCache: chunkCache,
260+
DailyBucketsFrom: model.TimeFromUnix(dailyBucketsFrom.Unix()),
253261
})
254262
if err != nil {
255263
return nil, err

0 commit comments

Comments
 (0)