Use ingester prefix for LifecyclerConfig

Many ingester flags were recently broken. This fixes the flags that were
broken, but if anyone is running the new HA Ruler they will need to
update their consul flags.

Signed-off-by: Chris Marchbanks <[email protected]>

Author: csmarchbanks

pkg/ingester/ingester.go

@@ -100,7 +100,7 @@ type Config struct {
 
 // RegisterFlags adds the flags required to config this to the given FlagSet
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
-	cfg.LifecyclerConfig.RegisterFlags(f)
+	cfg.LifecyclerConfig.RegisterFlagsWithPrefix("ingester.", f)
 
 	f.IntVar(&cfg.MaxTransferRetries, "ingester.max-transfer-retries", 10, "Number of times to try and transfer chunks before falling back to flushing.")
 	f.DurationVar(&cfg.FlushCheckPeriod, "ingester.flush-period", 1*time.Minute, "Period with which to attempt to flush chunks.")

pkg/ring/consul_client.go

@@ -13,6 +13,7 @@ import (
 	cleanhttp "github.com/hashicorp/go-cleanhttp"
 
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/weaveworks/common/httpgrpc"
 	"github.com/weaveworks/common/instrument"
 )
@@ -30,14 +31,14 @@ type ConsulConfig struct {
 	ConsistentReads   bool
 }
 
-// RegisterFlags adds the flags required to config this to the given FlagSet
+// RegisterFlags adds the flags required to config this to the given FlagSet.
 // If prefix is not an empty string it should end with a period.
 func (cfg *ConsulConfig) RegisterFlags(f *flag.FlagSet, prefix string) {
-	f.StringVar(&cfg.Host, prefix+"consul.hostname", "localhost:8500", "Hostname and port of Consul.")
-	f.StringVar(&cfg.Prefix, prefix+"consul.prefix", "collectors/", "Prefix for keys in Consul. Should end with a /.")
-	f.StringVar(&cfg.ACLToken, prefix+"consul.acltoken", "", "ACL Token used to interact with Consul.")
-	f.DurationVar(&cfg.HTTPClientTimeout, prefix+"consul.client-timeout", 2*longPollDuration, "HTTP timeout when talking to consul")
-	f.BoolVar(&cfg.ConsistentReads, prefix+"consul.consistent-reads", true, "Enable consistent reads to consul.")
+	flagext.StringVarOnce(f, &cfg.Host, prefix+"consul.hostname", "localhost:8500", "Hostname and port of Consul.")
+	flagext.StringVarOnce(f, &cfg.Prefix, prefix+"consul.prefix", "collectors/", "Prefix for keys in Consul. Should end with a /.")
+	flagext.StringVarOnce(f, &cfg.ACLToken, prefix+"consul.acltoken", "", "ACL Token used to interact with Consul.")
+	flagext.DurationVarOnce(f, &cfg.HTTPClientTimeout, prefix+"consul.client-timeout", 2*longPollDuration, "HTTP timeout when talking to consul")
+	flagext.BoolVarOnce(f, &cfg.ConsistentReads, prefix+"consul.consistent-reads", true, "Enable consistent reads to consul.")
 }
 
 type kv interface {

pkg/ring/kvstore.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/golang/protobuf/proto"
 	"github.com/golang/snappy"
 )
@@ -47,7 +48,7 @@ func (cfg *KVConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	if prefix == "" {
 		prefix = "ring."
 	}
-	f.StringVar(&cfg.Store, prefix+"store", "consul", "Backend storage to use for the ring (consul, inmemory).")
+	flagext.StringVarOnce(f, &cfg.Store, prefix+"store", "consul", "Backend storage to use for the ring (consul, inmemory).")
 }
 
 // CASCallback is the type of the callback to CAS.  If err is nil, out must be non-nil.

pkg/ring/lifecycler.go

@@ -41,7 +41,7 @@ var (
 type LifecyclerConfig struct {
 	RingConfig Config `yaml:"ring,omitempty"`
 
-	// Config for the ingester lifecycle control
+	// Config for the ingester lifecycle control.
 	ListenPort       *int
 	NumTokens        int           `yaml:"num_tokens,omitempty"`
 	HeartbeatPeriod  time.Duration `yaml:"heartbeat_period,omitempty"`
@@ -52,21 +52,21 @@ type LifecyclerConfig struct {
 	InfNames         []string      `yaml:"interface_names"`
 	FinalSleep       time.Duration `yaml:"final_sleep"`
 
-	// For testing, you can override the address and ID of this ingester
+	// For testing, you can override the address and ID of this ingester.
 	Addr           string `yaml:"address"`
 	Port           int
 	ID             string
 	SkipUnregister bool
 }
 
-// RegisterFlags adds the flags required to config this to the given FlagSet
+// RegisterFlags adds the flags required to config this to the given FlagSet.
 func (cfg *LifecyclerConfig) RegisterFlags(f *flag.FlagSet) {
 	cfg.RegisterFlagsWithPrefix("", f)
 }
 
-// RegisterFlagsWithPrefix adds the flags required to config this to the given FlagSet
+// RegisterFlagsWithPrefix adds the flags required to config this to the given FlagSet.
 func (cfg *LifecyclerConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
-	cfg.RingConfig.RegisterFlagsWithPrefix(prefix, f)
+	cfg.RingConfig.RegisterFlags(f)
 
 	f.IntVar(&cfg.NumTokens, prefix+"num-tokens", 128, "Number of tokens for each ingester.")
 	f.DurationVar(&cfg.HeartbeatPeriod, prefix+"heartbeat-period", 5*time.Second, "Period at which to heartbeat to consul.")

pkg/ring/ring.go

@@ -16,6 +16,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
 )
 
 const (
@@ -61,17 +62,12 @@ type Config struct {
 	ReplicationFactor int           `yaml:"replication_factor,omitempty"`
 }
 
-// RegisterFlags adds the flags required to config this to the given FlagSet with a specified prefix
+// RegisterFlags adds the flags required to config this to the given FlagSet.
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
-	cfg.RegisterFlagsWithPrefix("", f)
-}
-
-// RegisterFlagsWithPrefix adds the flags required to config this to the given FlagSet with a specified prefix
-func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
-	cfg.KVStore.RegisterFlagsWithPrefix(prefix, f)
+	cfg.KVStore.RegisterFlagsWithPrefix("", f)
 
-	f.DurationVar(&cfg.HeartbeatTimeout, prefix+"ring.heartbeat-timeout", time.Minute, "The heartbeat timeout after which ingesters are skipped for reads/writes.")
-	f.IntVar(&cfg.ReplicationFactor, prefix+"distributor.replication-factor", 3, "The number of ingesters to write to and read from.")
+	flagext.DurationVarOnce(f, &cfg.HeartbeatTimeout, "ring.heartbeat-timeout", time.Minute, "The heartbeat timeout after which ingesters are skipped for reads/writes.")
+	flagext.IntVarOnce(f, &cfg.ReplicationFactor, "distributor.replication-factor", 3, "The number of ingesters to write to and read from.")
 }
 
 // Ring holds the information about the members of the consistent hash ring.

pkg/util/flagext/once.go

@@ -0,0 +1,50 @@
+package flagext
+
+import (
+	"flag"
+	"time"
+)
+
+// IntVarOnce will check to see if a flag has already been registered. If
+// so it will set the value to the value already parsed. If not, it will
+// register the new flag.
+func IntVarOnce(f *flag.FlagSet, p *int, name string, value int, usage string) {
+	if fl := f.Lookup(name); fl == nil {
+		f.IntVar(p, name, value, usage)
+	} else {
+		*p = fl.Value.(flag.Getter).Get().(int)
+	}
+}
+
+// BoolVarOnce will check to see if a flag has already been registered. If
+// so it will set the value to the value already parsed. If not, it will
+// register the new flag.
+func BoolVarOnce(f *flag.FlagSet, p *bool, name string, value bool, usage string) {
+	if fl := f.Lookup(name); fl == nil {
+		f.BoolVar(p, name, value, usage)
+	} else {
+		*p = fl.Value.(flag.Getter).Get().(bool)
+	}
+}
+
+// StringVarOnce will check to see if a flag has already been registered. If
+// so it will set the value to the value already parsed. If not, it will
+// register the new flag.
+func StringVarOnce(f *flag.FlagSet, p *string, name string, value string, usage string) {
+	if fl := f.Lookup(name); fl == nil {
+		f.StringVar(p, name, value, usage)
+	} else {
+		*p = fl.Value.(flag.Getter).Get().(string)
+	}
+}
+
+// DurationVarOnce will check to see if a flag has already been registered. If
+// so it will set the value to the value already parsed. If not, it will
+// register the new flag.
+func DurationVarOnce(f *flag.FlagSet, p *time.Duration, name string, value time.Duration, usage string) {
+	if fl := f.Lookup(name); fl == nil {
+		f.DurationVar(p, name, value, usage)
+	} else {
+		*p = fl.Value.(flag.Getter).Get().(time.Duration)
+	}
+}