Skip to content

Commit 2be2a2e

Browse files
tomwilkietomwilkie
authored
Merge pull request #218 from weaveworks/151-split-it-up
Split up the single Cortex binary into cortex-{distributor, ingester, querier, ruler, table-manager}
2 parents ab0069d + 82e7dfe commit 2be2a2e

39 files changed

+877
-699
lines changed

‎.gitignore

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,8 @@
1-
cmd/cortex/cortex
2-
cmd/cortex_table_manager/cortex_table_manager
1+
cmd/distributor/distributor
2+
cmd/ingester/ingester
3+
cmd/querier/querier
4+
cmd/ruler/ruler
5+
cmd/table-manager/table-manager
36
.uptodate
47
.pkg
58
*.pb.go

‎Makefile

Lines changed: 29 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33

44
# Boiler plate for bulding Docker containers.
55
# All this must go at top of file I'm afraid.
6-
IMAGE_PREFIX := weaveworks
6+
IMAGE_PREFIX := weaveworks/cortex-
77
IMAGE_TAG := $(shell ./tools/image-tag)
88
UPTODATE := .uptodate
99

@@ -12,40 +12,43 @@ UPTODATE := .uptodate
1212
# Dependencies (i.e. things that go in the image) still need to be explicitly
1313
# declared.
1414
%/$(UPTODATE): %/Dockerfile
15-
$(SUDO) docker build -t $(IMAGE_PREFIX)/$(shell basename $(@D)) $(@D)/
16-
$(SUDO) docker tag $(IMAGE_PREFIX)/$(shell basename $(@D)) $(IMAGE_PREFIX)/$(shell basename $(@D)):$(IMAGE_TAG)
15+
$(SUDO) docker build -t $(IMAGE_PREFIX)$(shell basename $(@D)) $(@D)/
16+
$(SUDO) docker tag $(IMAGE_PREFIX)$(shell basename $(@D)) $(IMAGE_PREFIX)$(shell basename $(@D)):$(IMAGE_TAG)
1717
touch $@
1818

1919
# Get a list of directories containing Dockerfiles
2020
DOCKERFILES := $(shell find . -type f -name Dockerfile ! -path "./tools/*" ! -path "./vendor/*")
2121
UPTODATE_FILES := $(patsubst %/Dockerfile,%/$(UPTODATE),$(DOCKERFILES))
22-
DOCKER_IMAGE_DIRS=$(patsubst %/Dockerfile,%,$(DOCKERFILES))
23-
IMAGE_NAMES=$(foreach dir,$(DOCKER_IMAGE_DIRS),$(patsubst %,$(IMAGE_PREFIX)/%,$(shell basename $(dir))))
22+
DOCKER_IMAGE_DIRS := $(patsubst %/Dockerfile,%,$(DOCKERFILES))
23+
IMAGE_NAMES := $(foreach dir,$(DOCKER_IMAGE_DIRS),$(patsubst %,$(IMAGE_PREFIX)%,$(shell basename $(dir))))
2424
images:
2525
$(info $(IMAGE_NAMES))
26+
@echo > /dev/null
2627

28+
# Generating proto code is automated.
2729
PROTO_DEFS := $(shell find . -type f -name "*.proto" ! -path "./tools/*" ! -path "./vendor/*")
2830
PROTO_GOS := $(patsubst %.proto,%.pb.go,$(PROTO_DEFS))
2931

30-
# List of exes please
31-
CORTEX_EXE := ./cmd/cortex/cortex
32-
CORTEX_TABLE_MANAGER_EXE := ./cmd/cortex_table_manager/cortex_table_manager
33-
EXES = $(CORTEX_EXE) $(CORTEX_TABLE_MANAGER_EXE)
34-
32+
# Building binaries is now automated. The convention is to build a binary
33+
# for every directory with main.go in it, in the ./cmd directory.
34+
MAIN_GO := $(shell find ./cmd -type f -name main.go ! -path "./tools/*" ! -path "./vendor/*")
35+
EXES := $(foreach exe, $(patsubst ./cmd/%/main.go, %, $(MAIN_GO)), ./cmd/$(exe)/$(exe))
36+
GO_FILES := $(shell find . -name '*.go' ! -path "./cmd/*" ! -path "./tools/*" ! -path "./vendor/*")
37+
define dep_exe
38+
$(1): $(dir $(1))/main.go $(GO_FILES) ui/bindata.go $(PROTO_GOS)
39+
$(dir $(1))$(UPTODATE): $(1)
40+
endef
41+
$(foreach exe, $(EXES), $(eval $(call dep_exe, $(exe))))
42+
43+
# Manually declared dependancies And what goes into each exe
44+
%.pb.go: %.proto
3545
all: $(UPTODATE_FILES)
3646
test: $(PROTO_GOS)
37-
38-
# And what goes into each exe
39-
$(CORTEX_EXE): $(shell find . -name '*.go' ! -path "./tools/*" ! -path "./vendor/*") ui/bindata.go $(PROTO_GOS)
40-
$(CORTEX_TABLE_MANAGER_EXE): $(shell find ./chunk/ -name '*.go') cmd/cortex_table_manager/main.go
41-
%.pb.go: %.proto
4247
ui/bindata.go: $(shell find ui/static ui/templates)
4348
test: $(PROTO_GOS)
4449

4550
# And now what goes into each image
46-
cortex-build/$(UPTODATE): cortex-build/*
47-
cmd/cortex/$(UPTODATE): $(CORTEX_EXE)
48-
cmd/cortex_table_manager/$(UPTODATE): $(CORTEX_TABLE_MANAGER_EXE)
51+
build/$(UPTODATE): build/*
4952

5053
# All the boiler plate for building golang follows:
5154
SUDO := $(shell docker info >/dev/null 2>&1 || echo "sudo -E")
@@ -63,32 +66,32 @@ NETGO_CHECK = @strings $@ | grep cgo_stub\\\.go >/dev/null || { \
6366

6467
ifeq ($(BUILD_IN_CONTAINER),true)
6568

66-
$(EXES) $(PROTO_GOS) ui/bindata.go lint test shell: cortex-build/$(UPTODATE)
69+
$(EXES) $(PROTO_GOS) ui/bindata.go lint test shell: build/$(UPTODATE)
6770
@mkdir -p $(shell pwd)/.pkg
6871
$(SUDO) docker run $(RM) -ti \
6972
-v $(shell pwd)/.pkg:/go/pkg \
7073
-v $(shell pwd):/go/src/github.com/weaveworks/cortex \
71-
$(IMAGE_PREFIX)/cortex-build $@
74+
$(IMAGE_PREFIX)build $@
7275

7376
else
7477

75-
$(EXES): cortex-build/$(UPTODATE)
78+
$(EXES): build/$(UPTODATE)
7679
go build $(GO_FLAGS) -o $@ ./$(@D)
7780
$(NETGO_CHECK)
7881

79-
%.pb.go: cortex-build/$(UPTODATE)
82+
%.pb.go: build/$(UPTODATE)
8083
protoc -I ./vendor:./$(@D) --go_out=plugins=grpc:./$(@D) ./$(patsubst %.pb.go,%.proto,$@)
8184

82-
ui/bindata.go: cortex-build/$(UPTODATE)
85+
ui/bindata.go: build/$(UPTODATE)
8386
go-bindata -pkg ui -o ui/bindata.go -ignore '(.*\.map|bootstrap\.js|bootstrap-theme\.css|bootstrap\.css)' ui/templates/... ui/static/...
8487

85-
lint: cortex-build/$(UPTODATE)
88+
lint: build/$(UPTODATE)
8689
./tools/lint -notestpackage -ignorespelling queriers -ignorespelling Queriers .
8790

88-
test: cortex-build/$(UPTODATE)
91+
test: build/$(UPTODATE)
8992
./tools/test -netgo
9093

91-
shell: cortex-build/$(UPTODATE)
94+
shell: build/$(UPTODATE)
9295
bash
9396

9497
endif

‎cortex-build/Dockerfile renamed to ‎build/Dockerfile

File renamed without changes.

‎cortex-build/build.sh renamed to ‎build/build.sh

File renamed without changes.

‎chunk/chunk_cache.go

Lines changed: 38 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package chunk
22

33
import (
44
"bytes"
5+
"flag"
56
"fmt"
67
"io/ioutil"
78
"time"
@@ -47,10 +48,34 @@ type Memcache interface {
4748
Set(item *memcache.Item) error
4849
}
4950

51+
// CacheConfig is config to make a Cache
52+
type CacheConfig struct {
53+
Expiration time.Duration
54+
memcacheConfig MemcacheConfig
55+
}
56+
57+
// RegisterFlags adds the flags required to config this to the given FlagSet
58+
func (cfg *CacheConfig) RegisterFlags(f *flag.FlagSet) {
59+
f.DurationVar(&cfg.Expiration, "memcached.expiration", 0, "How long chunks stay in the memcache.")
60+
cfg.memcacheConfig.RegisterFlags(f)
61+
}
62+
5063
// Cache type caches chunks
5164
type Cache struct {
52-
Memcache Memcache
53-
Expiration time.Duration
65+
cfg CacheConfig
66+
memcache Memcache
67+
}
68+
69+
// NewCache makes a new Cache
70+
func NewCache(cfg CacheConfig) *Cache {
71+
var memcache Memcache
72+
if cfg.memcacheConfig.Host != "" {
73+
memcache = NewMemcacheClient(cfg.memcacheConfig)
74+
}
75+
return &Cache{
76+
cfg: cfg,
77+
memcache: memcache,
78+
}
5479
}
5580

5681
func memcacheStatusCode(err error) string {
@@ -73,6 +98,10 @@ func memcacheKey(userID, chunkID string) string {
7398

7499
// FetchChunkData gets chunks from the chunk cache.
75100
func (c *Cache) FetchChunkData(ctx context.Context, userID string, chunks []Chunk) (found []Chunk, missing []Chunk, err error) {
101+
if c.memcache == nil {
102+
return nil, chunks, nil
103+
}
104+
76105
memcacheRequests.Add(float64(len(chunks)))
77106

78107
keys := make([]string, 0, len(chunks))
@@ -83,7 +112,7 @@ func (c *Cache) FetchChunkData(ctx context.Context, userID string, chunks []Chun
83112
var items map[string]*memcache.Item
84113
err = instrument.TimeRequestHistogramStatus(ctx, "Memcache.Get", memcacheRequestDuration, memcacheStatusCode, func(_ context.Context) error {
85114
var err error
86-
items, err = c.Memcache.GetMulti(keys)
115+
items, err = c.memcache.GetMulti(keys)
87116
return err
88117
})
89118
if err != nil {
@@ -111,6 +140,10 @@ func (c *Cache) FetchChunkData(ctx context.Context, userID string, chunks []Chun
111140

112141
// StoreChunkData serializes and stores a chunk in the chunk cache.
113142
func (c *Cache) StoreChunkData(ctx context.Context, userID string, chunk *Chunk) error {
143+
if c.memcache == nil {
144+
return nil
145+
}
146+
114147
reader, err := chunk.reader()
115148
if err != nil {
116149
return err
@@ -125,9 +158,9 @@ func (c *Cache) StoreChunkData(ctx context.Context, userID string, chunk *Chunk)
125158
item := memcache.Item{
126159
Key: memcacheKey(userID, chunk.ID),
127160
Value: buf,
128-
Expiration: int32(c.Expiration.Seconds()),
161+
Expiration: int32(c.cfg.Expiration.Seconds()),
129162
}
130-
return c.Memcache.Set(&item)
163+
return c.memcache.Set(&item)
131164
})
132165
}
133166

‎chunk/chunk_store.go

Lines changed: 41 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"bytes"
55
"encoding/base64"
66
"encoding/json"
7+
"flag"
78
"fmt"
89
"sort"
910
"strconv"
@@ -21,6 +22,7 @@ import (
2122
"golang.org/x/net/context"
2223

2324
"github.com/weaveworks/cortex/user"
25+
"github.com/weaveworks/cortex/util"
2426
)
2527

2628
const (
@@ -89,20 +91,28 @@ type Store interface {
8991

9092
// StoreConfig specifies config for a ChunkStore
9193
type StoreConfig struct {
92-
S3 S3Client
93-
BucketName string
94-
DynamoDB DynamoDBClient
95-
TableName string
96-
ChunkCache *Cache
94+
PeriodicTableConfig
95+
CacheConfig
96+
S3 S3ClientValue
97+
DynamoDB DynamoDBClientValue
9798

9899
// After midnight on this day, we start bucketing indexes by day instead of by
99100
// hour. Only the day matters, not the time within the day.
100-
DailyBucketsFrom model.Time
101+
DailyBucketsFrom util.DayValue
101102

102103
// After this time, we will only query for base64-encoded label values.
103-
Base64ValuesFrom model.Time
104+
Base64ValuesFrom util.DayValue
105+
}
104106

105-
PeriodicTableConfig
107+
// RegisterFlags adds the flags required to config this to the given FlagSet
108+
func (cfg *StoreConfig) RegisterFlags(f *flag.FlagSet) {
109+
cfg.PeriodicTableConfig.RegisterFlags(f)
110+
cfg.CacheConfig.RegisterFlags(f)
111+
112+
f.Var(&cfg.S3, "s3.url", "S3 endpoint URL.")
113+
f.Var(&cfg.DynamoDB, "dynamodb.url", "DynamoDB endpoint URL.")
114+
f.Var(&cfg.DailyBucketsFrom, "dynamodb.daily-buckets-from", "The date in the format YYYY-MM-DD of the first day for which DynamoDB index buckets should be day-sized vs. hour-sized.")
115+
f.Var(&cfg.Base64ValuesFrom, "dynamodb.base64-buckets-from", "The date in the format YYYY-MM-DD after which we will stop querying to non-base64 encoded values.")
106116
}
107117

108118
// PeriodicTableConfig for the use of periodic tables (ie, weekly talbes). Can
@@ -112,20 +122,29 @@ type PeriodicTableConfig struct {
112122
UsePeriodicTables bool
113123
TablePrefix string
114124
TablePeriod time.Duration
115-
PeriodicTableStartAt time.Time
125+
PeriodicTableStartAt util.DayValue
126+
}
127+
128+
// RegisterFlags adds the flags required to config this to the given FlagSet
129+
func (cfg *PeriodicTableConfig) RegisterFlags(f *flag.FlagSet) {
130+
f.BoolVar(&cfg.UsePeriodicTables, "dynamodb.use-periodic-tables", true, "Should we user periodic tables.")
131+
f.StringVar(&cfg.TablePrefix, "dynamodb.periodic-table.prefix", "cortex_", "DynamoDB table prefix for the periodic tables.")
132+
f.DurationVar(&cfg.TablePeriod, "dynamodb.periodic-table.period", 7*24*time.Hour, "DynamoDB periodic tables period.")
133+
f.Var(&cfg.PeriodicTableStartAt, "dynamodb.periodic-table.start", "DynamoDB periodic tables start time.")
116134
}
117135

118136
// AWSStore implements ChunkStore for AWS
119137
type AWSStore struct {
120-
cfg StoreConfig
121-
138+
cfg StoreConfig
139+
cache *Cache
122140
dynamo *dynamoDBBackoffClient
123141
}
124142

125143
// NewAWSStore makes a new ChunkStore
126144
func NewAWSStore(cfg StoreConfig) *AWSStore {
127145
return &AWSStore{
128146
cfg: cfg,
147+
cache: NewCache(cfg.CacheConfig),
129148
dynamo: newDynamoDBBackoffClient(cfg.DynamoDB),
130149
}
131150
}
@@ -184,7 +203,7 @@ func (c *AWSStore) bigBuckets(from, through model.Time) []bucketSpec {
184203

185204
func (c *AWSStore) tableForBucket(bucketStart int64) string {
186205
if !c.cfg.UsePeriodicTables || bucketStart < (c.cfg.PeriodicTableStartAt.Unix()) {
187-
return c.cfg.TableName
206+
return c.cfg.DynamoDB.TableName
188207
}
189208
return c.cfg.TablePrefix + strconv.Itoa(int(bucketStart/int64(c.cfg.TablePeriod/time.Second)))
190209
}
@@ -356,7 +375,7 @@ func (c *AWSStore) putChunk(ctx context.Context, userID string, chunk *Chunk) er
356375
var err error
357376
_, err = c.cfg.S3.PutObject(&s3.PutObjectInput{
358377
Body: body,
359-
Bucket: aws.String(c.cfg.BucketName),
378+
Bucket: aws.String(c.cfg.S3.BucketName),
360379
Key: aws.String(chunkName(userID, chunk.ID)),
361380
})
362381
return err
@@ -365,10 +384,8 @@ func (c *AWSStore) putChunk(ctx context.Context, userID string, chunk *Chunk) er
365384
return err
366385
}
367386

368-
if c.cfg.ChunkCache != nil {
369-
if err = c.cfg.ChunkCache.StoreChunkData(ctx, userID, chunk); err != nil {
370-
log.Warnf("Could not store %v in chunk cache: %v", chunk.ID, err)
371-
}
387+
if err = c.cache.StoreChunkData(ctx, userID, chunk); err != nil {
388+
log.Warnf("Could not store %v in chunk cache: %v", chunk.ID, err)
372389
}
373390
return nil
374391
}
@@ -433,22 +450,18 @@ func (c *AWSStore) Get(ctx context.Context, from, through model.Time, matchers .
433450
queryChunks.Observe(float64(len(missing)))
434451

435452
var fromCache []Chunk
436-
if c.cfg.ChunkCache != nil {
437-
fromCache, missing, err = c.cfg.ChunkCache.FetchChunkData(ctx, userID, missing)
438-
if err != nil {
439-
log.Warnf("Error fetching from cache: %v", err)
440-
}
453+
fromCache, missing, err = c.cache.FetchChunkData(ctx, userID, missing)
454+
if err != nil {
455+
log.Warnf("Error fetching from cache: %v", err)
441456
}
442457

443458
fromS3, err := c.fetchChunkData(ctx, userID, missing)
444459
if err != nil {
445460
return nil, err
446461
}
447462

448-
if c.cfg.ChunkCache != nil {
449-
if err = c.cfg.ChunkCache.StoreChunks(ctx, userID, fromS3); err != nil {
450-
log.Warnf("Could not store chunks in chunk cache: %v", err)
451-
}
463+
if err = c.cache.StoreChunks(ctx, userID, fromS3); err != nil {
464+
log.Warnf("Could not store chunks in chunk cache: %v", err)
452465
}
453466

454467
// TODO instead of doing this sort, propagate an index and assign chunks
@@ -611,7 +624,7 @@ func (c *AWSStore) lookupChunksForMatcher(ctx context.Context, userID string, bu
611624
return nil, err
612625
}
613626

614-
if matcher.Type == metric.Equal && bucket.startTime.Before(c.cfg.Base64ValuesFrom) {
627+
if matcher.Type == metric.Equal && bucket.startTime.Before(c.cfg.Base64ValuesFrom.Time) {
615628
legacyRangePrefix, err := rangeValueKeyAndValueOnly(matcher.Name, matcher.Value)
616629
if err != nil {
617630
return nil, err
@@ -715,7 +728,7 @@ func (c *AWSStore) fetchChunkData(ctx context.Context, userID string, chunkSet [
715728
err := instrument.TimeRequestHistogram(ctx, "S3.GetObject", s3RequestDuration, func(_ context.Context) error {
716729
var err error
717730
resp, err = c.cfg.S3.GetObject(&s3.GetObjectInput{
718-
Bucket: aws.String(c.cfg.BucketName),
731+
Bucket: aws.String(c.cfg.S3.BucketName),
719732
Key: aws.String(chunkName(userID, chunk.ID)),
720733
})
721734
return err

0 commit comments

Comments
 (0)