Merge pull request #165 from weaveworks/backout-protoring

Revert "Merge pull request #159 from weaveworks/protoring"

Author: tomwilkie

.gitignore

@@ -1,5 +1,5 @@
 cmd/cortex/cortex
 .uptodate
 .pkg
-*.pb.go
+cortex.pb.go
 ui/bindata.go

Makefile

@@ -24,19 +24,15 @@ IMAGE_NAMES=$(foreach dir,$(DOCKER_IMAGE_DIRS),$(patsubst %,$(IMAGE_PREFIX)/%,$(
 images:
 	$(info $(IMAGE_NAMES))
 
-PROTO_DEFS := $(shell find . -type f -name "*.proto" ! -path "./tools/*" ! -path "./vendor/*")
-PROTO_GOS := $(patsubst %.proto,%.pb.go,$(PROTO_DEFS))
-
 # List of exes please
 CORTEX_EXE := ./cmd/cortex/cortex
 EXES = $(CORTEX_EXE)
 
 all: $(UPTODATE_FILES)
-test: $(PROTO_GOS)
 
 # And what goes into each exe
-$(CORTEX_EXE): $(shell find . -name '*.go' ! -path "./tools/*" ! -path "./vendor/*") ui/bindata.go $(PROTO_GOS)
-%.pb.go: %.proto
+$(CORTEX_EXE): $(shell find . -name '*.go') ui/bindata.go cortex.pb.go
+cortex.pb.go: cortex.proto
 ui/bindata.go: $(shell find ui/static ui/templates)
 
 # And now what goes into each image
@@ -59,7 +55,7 @@ NETGO_CHECK = @strings $@ | grep cgo_stub\\\.go >/dev/null || { \
 
 ifeq ($(BUILD_IN_CONTAINER),true)
 
-$(EXES) $(PROTO_GOS) ui/bindata.go lint test shell: cortex-build/$(UPTODATE)
+$(EXES) cortex.pb.go ui/bindata.go lint test shell: cortex-build/$(UPTODATE)
 	@mkdir -p $(shell pwd)/.pkg
 	$(SUDO) docker run $(RM) -ti \
 		-v $(shell pwd)/.pkg:/go/pkg \
@@ -72,16 +68,16 @@ $(EXES): cortex-build/$(UPTODATE)
 	go build $(GO_FLAGS) -o $@ ./$(@D)
 	$(NETGO_CHECK)
 
-%.pb.go: cortex-build/$(UPTODATE)
-	protoc -I ./vendor:./$(@D) --go_out=plugins=grpc:./$(@D) ./$(patsubst %.pb.go,%.proto,$@)
+cortex.pb.go: cortex-build/$(UPTODATE)
+	protoc -I ./vendor:. --go_out=plugins=grpc:. ./cortex.proto
 
 ui/bindata.go: cortex-build/$(UPTODATE)
 	go-bindata -pkg ui -o ui/bindata.go -ignore '(.*\.map|bootstrap\.js|bootstrap-theme\.css|bootstrap\.css)'  ui/templates/... ui/static/...
 
 lint: cortex-build/$(UPTODATE)
 	./tools/lint -notestpackage -ignorespelling queriers -ignorespelling Queriers .
 
-test: cortex-build/$(UPTODATE)
+test: cortex-build/$(UPTODATE) cortex.pb.go
 	./tools/test -no-go-get
 
 shell: cortex-build/$(UPTODATE)
@@ -91,7 +87,7 @@ endif
 
 clean:
 	$(SUDO) docker rmi $(IMAGE_NAMES) >/dev/null 2>&1 || true
-	rm -rf $(UPTODATE_FILES) $(EXES) $(PROTO_GOS) ui/bindata.go
+	rm -rf $(UPTODATE_FILES) $(EXES) cortex.pb.go ui/bindata.go
 	go clean ./...
 
 

cmd/cortex/main.go

@@ -138,11 +138,7 @@ func main() {
 		prometheus.MustRegister(resourceWatcher)
 	}
 
-	ringCodec := ring.NewDynamicCodec(
-		ring.JSONCodec{Factory: ring.DescFactory},
-		ring.ProtoCodec{Factory: ring.ProtoDescFactory},
-	)
-	consul, err := ring.NewConsulClient(cfg.consulHost, ringCodec)
+	consul, err := ring.NewConsulClient(cfg.consulHost)
 	if err != nil {
 		log.Fatalf("Error initializing Consul client: %v", err)
 	}
@@ -160,7 +156,7 @@ func main() {
 
 	case modeIngester:
 		cfg.ingesterConfig.Ring = r
-		registration, err := ring.RegisterIngester(consul, cfg.ingesterConfig.GRPCListenPort, cfg.numTokens, ringCodec)
+		registration, err := ring.RegisterIngester(consul, cfg.listenPort, cfg.ingesterConfig.GRPCListenPort, cfg.numTokens)
 		if err != nil {
 			// This only happens for errors in configuration & set-up, not for
 			// network errors.
@@ -187,7 +183,7 @@ func main() {
 
 		// Deferring a func to make ordering obvious
 		defer func() {
-			registration.ChangeState(ring.IngesterState_LEAVING)
+			registration.ChangeState(ring.Leaving)
 			ing.Stop()
 			registration.Unregister()
 		}()

distributor/distributor.go

@@ -54,9 +54,9 @@ type Distributor struct {
 type ReadRing interface {
 	prometheus.Collector
 
-	Get(key uint32, n int, op ring.Operation) ([]*ring.IngesterDesc, error)
-	BatchGet(keys []uint32, n int, op ring.Operation) ([][]*ring.IngesterDesc, error)
-	GetAll() []*ring.IngesterDesc
+	Get(key uint32, n int, op ring.Operation) ([]ring.IngesterDesc, error)
+	BatchGet(keys []uint32, n int, op ring.Operation) ([][]ring.IngesterDesc, error)
+	GetAll() []ring.IngesterDesc
 }
 
 // Config contains the configuration require to
@@ -121,7 +121,7 @@ func New(cfg Config) (*Distributor, error) {
 	}, nil
 }
 
-func (d *Distributor) getClientFor(ingester *ring.IngesterDesc) (cortex.IngesterClient, error) {
+func (d *Distributor) getClientFor(ingester ring.IngesterDesc) (cortex.IngesterClient, error) {
 	d.clientsMtx.RLock()
 	client, ok := d.clients[ingester.Hostname]
 	d.clientsMtx.RUnlock()
@@ -137,7 +137,7 @@ func (d *Distributor) getClientFor(ingester *ring.IngesterDesc) (cortex.Ingester
 	}
 
 	conn, err := grpc.Dial(
-		ingester.Hostname,
+		ingester.GRPCHostname,
 		grpc.WithInsecure(),
 		grpc.WithUnaryInterceptor(grpc_middleware.ChainUnaryClient(
 			otgrpc.OpenTracingClientInterceptor(opentracing.GlobalTracer()),
@@ -192,7 +192,7 @@ func (d *Distributor) Push(ctx context.Context, req *remote.WriteRequest) (*cort
 	}
 
 	sampleTrackers := make([]sampleTracker, len(samples), len(samples))
-	samplesByIngester := map[*ring.IngesterDesc][]*sampleTracker{}
+	samplesByIngester := map[ring.IngesterDesc][]*sampleTracker{}
 	for i := range samples {
 		sampleTrackers[i] = sampleTracker{
 			sample: samples[i],
@@ -204,9 +204,9 @@ func (d *Distributor) Push(ctx context.Context, req *remote.WriteRequest) (*cort
 		// Skip those that have not heartbeated in a while. NB these are still
 		// included in the calculation of minSuccess, so if too many failed ingesters
 		// will cause the whole write to fail.
-		liveIngesters := make([]*ring.IngesterDesc, 0, len(ingesters[i]))
+		liveIngesters := make([]ring.IngesterDesc, 0, len(ingesters[i]))
 		for _, ingester := range ingesters[i] {
-			if time.Now().Sub(time.Unix(ingester.Timestamp, 0)) <= d.cfg.HeartbeatTimeout {
+			if time.Now().Sub(ingester.Timestamp) <= d.cfg.HeartbeatTimeout {
 				liveIngesters = append(liveIngesters, ingester)
 			}
 		}
@@ -226,7 +226,7 @@ func (d *Distributor) Push(ctx context.Context, req *remote.WriteRequest) (*cort
 
 	errs := make(chan error)
 	for hostname, samples := range samplesByIngester {
-		go func(ingester *ring.IngesterDesc, samples []*sampleTracker) {
+		go func(ingester ring.IngesterDesc, samples []*sampleTracker) {
 			errs <- d.sendSamples(ctx, ingester, samples)
 		}(hostname, samples)
 	}
@@ -246,7 +246,7 @@ func (d *Distributor) Push(ctx context.Context, req *remote.WriteRequest) (*cort
 	return &cortex.WriteResponse{}, nil
 }
 
-func (d *Distributor) sendSamples(ctx context.Context, ingester *ring.IngesterDesc, sampleTrackers []*sampleTracker) error {
+func (d *Distributor) sendSamples(ctx context.Context, ingester ring.IngesterDesc, sampleTrackers []*sampleTracker) error {
 	client, err := d.getClientFor(ingester)
 	if err != nil {
 		return err