Add a flag to control Cassandra consistency level (#700)

Signed-off-by: Yuri Shkuro <ys@uber.com>

Author: yurishkuro

plugin/storage/cassandra/options.go

@@ -33,6 +33,7 @@ const (
 	suffixServers          = ".servers"
 	suffixPort             = ".port"
 	suffixKeyspace         = ".keyspace"
+	suffixConsistency      = ".consistency"
 	suffixProtoVer         = ".proto-version"
 	suffixSocketKeepAlive  = ".socket-keep-alive"
 	suffixUsername         = ".username"
@@ -147,6 +148,10 @@ func addFlags(flagSet *flag.FlagSet, nsConfig *namespaceConfig) {
 		nsConfig.namespace+suffixKeyspace,
 		nsConfig.Keyspace,
 		"The Cassandra keyspace for Jaeger data")
+	flagSet.String(
+		nsConfig.namespace+suffixConsistency,
+		nsConfig.Consistency,
+		"The Cassandra consistency level, e.g. ANY, ONE, TWO, THREE, QUORUM, ALL, LOCAL_QUORUM, EACH_QUORUM, LOCAL_ONE (default LOCAL_ONE)")
 	flagSet.Int(
 		nsConfig.namespace+suffixProtoVer,
 		nsConfig.ProtoVersion,
@@ -209,6 +214,7 @@ func (cfg *namespaceConfig) initFromViper(v *viper.Viper) {
 	cfg.servers = v.GetString(cfg.namespace + suffixServers)
 	cfg.Port = v.GetInt(cfg.namespace + suffixPort)
 	cfg.Keyspace = v.GetString(cfg.namespace + suffixKeyspace)
+	cfg.Consistency = v.GetString(cfg.namespace + suffixConsistency)
 	cfg.ProtoVersion = v.GetInt(cfg.namespace + suffixProtoVer)
 	cfg.SocketKeepAlive = v.GetDuration(cfg.namespace + suffixSocketKeepAlive)
 	cfg.Authenticator.Basic.Username = v.GetString(cfg.namespace + suffixUsername)

plugin/storage/cassandra/options_test.go

@@ -53,6 +53,7 @@ func TestOptionsWithFlags(t *testing.T) {
 		"--cas.max-retry-attempts=42",
 		"--cas.timeout=42s",
 		"--cas.port=4242",
+		"--cas.consistency=ONE",
 		"--cas.proto-version=3",
 		"--cas.socket-keep-alive=42s",
 		// enable aux with a couple overrides
@@ -65,6 +66,7 @@ func TestOptionsWithFlags(t *testing.T) {
 	primary := opts.GetPrimary()
 	assert.Equal(t, "jaeger", primary.Keyspace)
 	assert.Equal(t, []string{"1.1.1.1", "2.2.2.2"}, primary.Servers)
+	assert.Equal(t, "ONE", primary.Consistency)
 
 	aux := opts.Get("cas-aux")
 	require.NotNil(t, aux)
@@ -74,6 +76,7 @@ func TestOptionsWithFlags(t *testing.T) {
 	assert.Equal(t, 42, aux.MaxRetryAttempts)
 	assert.Equal(t, 42*time.Second, aux.Timeout)
 	assert.Equal(t, 4242, aux.Port)
+	assert.Equal(t, "", aux.Consistency, "aux storage does not inherit consistency from primary")
 	assert.Equal(t, 3, aux.ProtoVersion)
 	assert.Equal(t, 42*time.Second, aux.SocketKeepAlive)
 }