Review feedback

Author: tomwilkie

cmd/cortex/main.go

@@ -93,7 +93,6 @@ func main() {
 	flag.IntVar(&cfg.numTokens, "ingester.num-tokens", 128, "Number of tokens for each ingester.")
 	flag.IntVar(&cfg.distributorConfig.ReplicationFactor, "distributor.replication-factor", 3, "The number of ingesters to write to and read from.")
 	flag.IntVar(&cfg.distributorConfig.MinReadSuccesses, "distributor.min-read-successes", 2, "The minimum number of ingesters from which a read must succeed.")
-	flag.IntVar(&cfg.distributorConfig.MinWriteSuccesses, "distributor.min-write-successes", 2, "The minimum number of ingesters to which a write must succeed.")
 	flag.DurationVar(&cfg.distributorConfig.HeartbeatTimeout, "distributor.heartbeat-timeout", time.Minute, "The heartbeat timeout after which ingesters are skipped for reads/writes.")
 	flag.BoolVar(&cfg.logSuccess, "log.success", false, "Log successful requests")
 	flag.Parse()
@@ -125,7 +124,6 @@ func main() {
 			// network errors.
 			log.Fatalf("Could not register ingester: %v", err)
 		}
-		prometheus.MustRegister(registration)
 		ing := setupIngester(chunkStore, cfg.ingesterConfig, cfg.logSuccess)
 
 		// Deferring a func to make ordering obvious
@@ -134,6 +132,8 @@ func main() {
 			ing.Stop()
 			registration.Unregister()
 		}()
+
+		prometheus.MustRegister(registration)
 	default:
 		log.Fatalf("Mode %s not supported!", cfg.mode)
 	}

distributor.go

@@ -63,7 +63,6 @@ type DistributorConfig struct {
 
 	ReplicationFactor int
 	MinReadSuccesses  int
-	MinWriteSuccesses int
 	HeartbeatTimeout  time.Duration
 }
 
@@ -72,9 +71,6 @@ func NewDistributor(cfg DistributorConfig) (*Distributor, error) {
 	if 0 > cfg.ReplicationFactor {
 		return nil, fmt.Errorf("ReplicationFactor must be greater than zero: %d", cfg.ReplicationFactor)
 	}
-	if cfg.MinWriteSuccesses > cfg.ReplicationFactor {
-		return nil, fmt.Errorf("MinWriteSuccesses > ReplicationFactor: %d > %d", cfg.MinWriteSuccesses, cfg.ReplicationFactor)
-	}
 	if cfg.MinReadSuccesses > cfg.ReplicationFactor {
 		return nil, fmt.Errorf("MinReadSuccesses > ReplicationFactor: %d > %d", cfg.MinReadSuccesses, cfg.ReplicationFactor)
 	}
@@ -171,10 +167,7 @@ func (d *Distributor) Append(ctx context.Context, samples []*model.Sample) error
 
 	d.receivedSamples.Add(float64(len(samples)))
 
-	sampleTrackers := make([]sampleTracker, len(samples), len(samples))
-	samplesByIngester := map[string][]*sampleTracker{}
 	keys := make([]uint32, len(samples), len(samples))
-
 	for i, sample := range samples {
 		keys[i] = tokenForMetric(userID, sample.Metric)
 	}
@@ -184,22 +177,28 @@ func (d *Distributor) Append(ctx context.Context, samples []*model.Sample) error
 		return err
 	}
 
+	sampleTrackers := make([]sampleTracker, len(samples), len(samples))
+	samplesByIngester := map[string][]*sampleTracker{}
 	for i := range samples {
 		sampleTrackers[i] = sampleTracker{
-			sample:     samples[i],
+			sample: samples[i],
+			// We need a response from a quorum of ingesters, which is n/2 + 1.
 			minSuccess: (len(ingesters[i]) / 2) + 1,
 			succeeded:  0,
 		}
 
 		// 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([]string, len(ingesters[i]))
+		liveIngesters := make([]string, 0, len(ingesters[i]))
 		for _, ingester := range ingesters[i] {
 			if time.Now().Sub(ingester.Timestamp) <= d.cfg.HeartbeatTimeout {
 				liveIngesters = append(liveIngesters, ingester.Hostname)
 			}
 		}
+
+		// This is just a shortcut - if there are not minSuccess available ingesters,
+		// after filtering out dead ones, don't even both trying.
 		if len(liveIngesters) < sampleTrackers[i].minSuccess {
 			return fmt.Errorf("wanted at least %d live ingesters to process write, had %d",
 				sampleTrackers[i].minSuccess, len(liveIngesters))

ring/ingester_lifecycle.go

@@ -32,7 +32,7 @@ type IngesterRegistration struct {
 	wait     sync.WaitGroup
 
 	// We need to remember the token state just in case consul goes away and comes
-	// back empty.  Channel is users to tell the actor to update consul on state changes.
+	// back empty.  Channel is used to tell the actor to update consul on state changes.
 	state       TokenState
 	stateChange chan TokenState
 
@@ -134,7 +134,7 @@ func (r *IngesterRegistration) pickTokens() []uint32 {
 }
 
 func (r *IngesterRegistration) heartbeat(tokens []uint32) {
-	heartbeat := func(in interface{}) (out interface{}, retry bool, err error) {
+	updateConsul := func(in interface{}) (out interface{}, retry bool, err error) {
 		var ringDesc *Desc
 		if in == nil {
 			ringDesc = newDesc()
@@ -163,12 +163,12 @@ func (r *IngesterRegistration) heartbeat(tokens []uint32) {
 	for {
 		select {
 		case r.state = <-r.stateChange:
-			if err := r.consul.CAS(consulKey, descFactory, heartbeat); err != nil {
+			if err := r.consul.CAS(consulKey, descFactory, updateConsul); err != nil {
 				log.Errorf("Failed to write to consul, sleeping: %v", err)
 			}
 		case <-ticker.C:
 			r.consulHeartbeats.Inc()
-			if err := r.consul.CAS(consulKey, descFactory, heartbeat); err != nil {
+			if err := r.consul.CAS(consulKey, descFactory, updateConsul); err != nil {
 				log.Errorf("Failed to write to consul, sleeping: %v", err)
 			}
 		case <-r.quit:

ring/ring.go

@@ -152,8 +152,12 @@ func (r *Ring) getInternal(key uint32, n int, op Operation) ([]IngesterDesc, err
 		}
 		distinctHosts[token.Ingester] = struct{}{}
 
-		// If we encounter a Leaving token, for reads we should bump n,
-		// for writes we bump n and skip the token.
+		// Ingesters that are Leaving do not count to the replication limit. We do
+		// not want to Write to them because they are about to go away, but we do
+		// want to write the extra replica somewhere.  So we increase the size of the
+		// set of replicas for the key.  This means we have to also increase the
+		// size of the replica set for read, but we can read from Leaving ingesters,
+		// so don't skip it in this case.
 		if token.State == Leaving {
 			n++
 			if op == Write {