Merge pull request #1407 from cortexproject/1342-cassandra-config-yaml

Allow Cassandra config to be passed as YAML.

Author: khaines

pkg/chunk/cassandra/fixtures.go

@@ -41,10 +41,10 @@ func Fixtures() ([]testutils.Fixture, error) {
 	}
 
 	cfg := Config{
-		addresses:         addresses,
-		keyspace:          "test",
-		consistency:       "QUORUM",
-		replicationFactor: 1,
+		Addresses:         addresses,
+		Keyspace:          "test",
+		Consistency:       "QUORUM",
+		ReplicationFactor: 1,
 	}
 
 	// Get a SchemaConfig with the defaults.

pkg/chunk/cassandra/storage_client.go

@@ -20,40 +20,40 @@ const (
 
 // Config for a StorageClient
 type Config struct {
-	addresses                string
-	port                     int
-	keyspace                 string
-	consistency              string
-	replicationFactor        int
-	disableInitialHostLookup bool
-	ssl                      bool
-	hostVerification         bool
-	caPath                   string
-	auth                     bool
-	username                 string
-	password                 string
-	timeout                  time.Duration
+	Addresses                string        `yaml:"addresses,omitempty"`
+	Port                     int           `yaml:"port,omitempty"`
+	Keyspace                 string        `yaml:"keyspace,omitempty"`
+	Consistency              string        `yaml:"consistency,omitempty"`
+	ReplicationFactor        int           `yaml:"replication_factor,omitempty"`
+	DisableInitialHostLookup bool          `yaml:"disable_initial_host_lookup,omitempty"`
+	SSL                      bool          `yaml:"SSL,omitempty"`
+	HostVerification         bool          `yaml:"host_verification,omitempty"`
+	CAPath                   string        `yaml:"CA_path,omitempty"`
+	Auth                     bool          `yaml:"auth,omitempty"`
+	Username                 string        `yaml:"username,omitempty"`
+	Password                 string        `yaml:"password,omitempty"`
+	Timeout                  time.Duration `yaml:"timeout,omitempty"`
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
-	f.StringVar(&cfg.addresses, "cassandra.addresses", "", "Comma-separated hostnames or ips of Cassandra instances.")
-	f.IntVar(&cfg.port, "cassandra.port", 9042, "Port that Cassandra is running on")
-	f.StringVar(&cfg.keyspace, "cassandra.keyspace", "", "Keyspace to use in Cassandra.")
-	f.StringVar(&cfg.consistency, "cassandra.consistency", "QUORUM", "Consistency level for Cassandra.")
-	f.IntVar(&cfg.replicationFactor, "cassandra.replication-factor", 1, "Replication factor to use in Cassandra.")
-	f.BoolVar(&cfg.disableInitialHostLookup, "cassandra.disable-initial-host-lookup", false, "Instruct the cassandra driver to not attempt to get host info from the system.peers table.")
-	f.BoolVar(&cfg.ssl, "cassandra.ssl", false, "Use SSL when connecting to cassandra instances.")
-	f.BoolVar(&cfg.hostVerification, "cassandra.host-verification", true, "Require SSL certificate validation.")
-	f.StringVar(&cfg.caPath, "cassandra.ca-path", "", "Path to certificate file to verify the peer.")
-	f.BoolVar(&cfg.auth, "cassandra.auth", false, "Enable password authentication when connecting to cassandra.")
-	f.StringVar(&cfg.username, "cassandra.username", "", "Username to use when connecting to cassandra.")
-	f.StringVar(&cfg.password, "cassandra.password", "", "Password to use when connecting to cassandra.")
-	f.DurationVar(&cfg.timeout, "cassandra.timeout", 600*time.Millisecond, "Timeout when connecting to cassandra.")
+	f.StringVar(&cfg.Addresses, "cassandra.addresses", "", "Comma-separated hostnames or IPs of Cassandra instances.")
+	f.IntVar(&cfg.Port, "cassandra.port", 9042, "Port that Cassandra is running on")
+	f.StringVar(&cfg.Keyspace, "cassandra.keyspace", "", "Keyspace to use in Cassandra.")
+	f.StringVar(&cfg.Consistency, "cassandra.consistency", "QUORUM", "Consistency level for Cassandra.")
+	f.IntVar(&cfg.ReplicationFactor, "cassandra.replication-factor", 1, "Replication factor to use in Cassandra.")
+	f.BoolVar(&cfg.DisableInitialHostLookup, "cassandra.disable-initial-host-lookup", false, "Instruct the cassandra driver to not attempt to get host info from the system.peers table.")
+	f.BoolVar(&cfg.SSL, "cassandra.ssl", false, "Use SSL when connecting to cassandra instances.")
+	f.BoolVar(&cfg.HostVerification, "cassandra.host-verification", true, "Require SSL certificate validation.")
+	f.StringVar(&cfg.CAPath, "cassandra.ca-path", "", "Path to certificate file to verify the peer.")
+	f.BoolVar(&cfg.Auth, "cassandra.auth", false, "Enable password authentication when connecting to cassandra.")
+	f.StringVar(&cfg.Username, "cassandra.username", "", "Username to use when connecting to cassandra.")
+	f.StringVar(&cfg.Password, "cassandra.password", "", "Password to use when connecting to cassandra.")
+	f.DurationVar(&cfg.Timeout, "cassandra.timeout", 600*time.Millisecond, "Timeout when connecting to cassandra.")
 }
 
 func (cfg *Config) session() (*gocql.Session, error) {
-	consistency, err := gocql.ParseConsistencyWrapper(cfg.consistency)
+	consistency, err := gocql.ParseConsistencyWrapper(cfg.Consistency)
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -62,40 +62,40 @@ func (cfg *Config) session() (*gocql.Session, error) {
 		return nil, errors.WithStack(err)
 	}
 
-	cluster := gocql.NewCluster(strings.Split(cfg.addresses, ",")...)
-	cluster.Port = cfg.port
-	cluster.Keyspace = cfg.keyspace
+	cluster := gocql.NewCluster(strings.Split(cfg.Addresses, ",")...)
+	cluster.Port = cfg.Port
+	cluster.Keyspace = cfg.Keyspace
 	cluster.Consistency = consistency
 	cluster.BatchObserver = observer{}
 	cluster.QueryObserver = observer{}
-	cluster.Timeout = cfg.timeout
+	cluster.Timeout = cfg.Timeout
 	cfg.setClusterConfig(cluster)
 
 	return cluster.CreateSession()
 }
 
 // apply config settings to a cassandra ClusterConfig
 func (cfg *Config) setClusterConfig(cluster *gocql.ClusterConfig) {
-	cluster.DisableInitialHostLookup = cfg.disableInitialHostLookup
+	cluster.DisableInitialHostLookup = cfg.DisableInitialHostLookup
 
-	if cfg.ssl {
+	if cfg.SSL {
 		cluster.SslOpts = &gocql.SslOptions{
-			CaPath:                 cfg.caPath,
-			EnableHostVerification: cfg.hostVerification,
+			CaPath:                 cfg.CAPath,
+			EnableHostVerification: cfg.HostVerification,
 		}
 	}
-	if cfg.auth {
+	if cfg.Auth {
 		cluster.Authenticator = gocql.PasswordAuthenticator{
-			Username: cfg.username,
-			Password: cfg.password,
+			Username: cfg.Username,
+			Password: cfg.Password,
 		}
 	}
 }
 
 // createKeyspace will create the desired keyspace if it doesn't exist.
 func (cfg *Config) createKeyspace() error {
-	cluster := gocql.NewCluster(strings.Split(cfg.addresses, ",")...)
-	cluster.Port = cfg.port
+	cluster := gocql.NewCluster(strings.Split(cfg.Addresses, ",")...)
+	cluster.Port = cfg.Port
 	cluster.Keyspace = "system"
 	cluster.Timeout = 20 * time.Second
 
@@ -113,7 +113,7 @@ func (cfg *Config) createKeyspace() error {
 			 'class' : 'SimpleStrategy',
 			 'replication_factor' : %d
 		 }`,
-		cfg.keyspace, cfg.replicationFactor)).Exec()
+		cfg.Keyspace, cfg.ReplicationFactor)).Exec()
 	return errors.WithStack(err)
 }
 

pkg/chunk/cassandra/table_client.go

@@ -28,7 +28,7 @@ func NewTableClient(ctx context.Context, cfg Config) (chunk.TableClient, error)
 }
 
 func (c *tableClient) ListTables(ctx context.Context) ([]string, error) {
-	md, err := c.session.KeyspaceMetadata(c.cfg.keyspace)
+	md, err := c.session.KeyspaceMetadata(c.cfg.Keyspace)
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}