backwards compatible cassandra timeout

Author: replay

cmd/mt-split-metrics-by-ttl/main.go

@@ -23,7 +23,7 @@ func main() {
 	flag.StringVar(&storeConfig.Keyspace, "cassandra-keyspace", storeConfig.Keyspace, "cassandra keyspace to use for storing the metric data table")
 	flag.StringVar(&storeConfig.Consistency, "cassandra-consistency", storeConfig.Consistency, "write consistency (any|one|two|three|quorum|all|local_quorum|each_quorum|local_one")
 	flag.StringVar(&storeConfig.HostSelectionPolicy, "cassandra-host-selection-policy", storeConfig.HostSelectionPolicy, "")
-	flag.DurationVar(&storeConfig.Timeout, "cassandra-timeout", storeConfig.Timeout, "cassandra timeout")
+	flag.StringVar(&storeConfig.Timeout, "cassandra-timeout", storeConfig.Timeout, "cassandra timeout")
 	flag.IntVar(&storeConfig.Retries, "cassandra-retries", storeConfig.Retries, "how many times to retry a query before failing it")
 	flag.IntVar(&storeConfig.CqlProtocolVersion, "cql-protocol-version", storeConfig.CqlProtocolVersion, "cql protocol version to use")
 	flag.BoolVar(&storeConfig.DisableInitialHostLookup, "cassandra-disable-initial-host-lookup", storeConfig.DisableInitialHostLookup, "instruct the driver to not attempt to get host info from the system.peers table")

cmd/mt-store-cat/main.go

@@ -45,7 +45,7 @@ func main() {
 	flag.StringVar(&storeConfig.Keyspace, "cassandra-keyspace", storeConfig.Keyspace, "cassandra keyspace to use for storing the metric data table")
 	flag.StringVar(&storeConfig.Consistency, "cassandra-consistency", storeConfig.Consistency, "write consistency (any|one|two|three|quorum|all|local_quorum|each_quorum|local_one")
 	flag.StringVar(&storeConfig.HostSelectionPolicy, "cassandra-host-selection-policy", storeConfig.HostSelectionPolicy, "")
-	flag.DurationVar(&storeConfig.Timeout, "cassandra-timeout", storeConfig.Timeout, "cassandra timeout")
+	flag.StringVar(&storeConfig.Timeout, "cassandra-timeout", storeConfig.Timeout, "cassandra timeout")
 	flag.IntVar(&storeConfig.ReadConcurrency, "cassandra-read-concurrency", storeConfig.ReadConcurrency, "max number of concurrent reads to cassandra.")
 	//flag.IntVar(&storeConfig.WriteConcurrency, "write-concurrency", storeConfig.WriteConcurrency, "max number of concurrent writes to cassandra.")
 	flag.IntVar(&storeConfig.ReadQueueSize, "cassandra-read-queue-size", storeConfig.ReadQueueSize, "max number of outstanding reads before reads will be dropped. This is important if you run queries that result in many reads in parallel.")

cmd/mt-store-cp-experimental/main.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/gocql/gocql"
+	"github.com/grafana/metrictank/store/cassandra"
 	hostpool "github.com/hailocab/go-hostpool"
 )
 
@@ -25,7 +26,7 @@ var (
 	cassandraKeyspace            = flag.String("cassandra-keyspace", "metrictank", "cassandra keyspace to use for storing the metric data table")
 	cassandraConsistency         = flag.String("cassandra-consistency", "one", "write consistency (any|one|two|three|quorum|all|local_quorum|each_quorum|local_one")
 	cassandraHostSelectionPolicy = flag.String("cassandra-host-selection-policy", "tokenaware,hostpool-epsilon-greedy", "")
-	cassandraTimeout             = flag.Int("cassandra-timeout", 1000, "cassandra timeout in milliseconds")
+	cassandraTimeout             = flag.String("cassandra-timeout", "1s", "cassandra timeout")
 	cassandraConcurrency         = flag.Int("cassandra-concurrency", 20, "max number of concurrent reads to cassandra.")
 	cassandraRetries             = flag.Int("cassandra-retries", 0, "how many times to retry a query before failing it")
 	cassandraDisableHostLookup   = flag.Bool("cassandra-disable-host-lookup", false, "disable host lookup (useful if going through proxy)")
@@ -119,7 +120,7 @@ func NewCassandraStore(cassandraAddrs *string) (*gocql.Session, error) {
 	}
 	cluster.DisableInitialHostLookup = *cassandraDisableHostLookup
 	cluster.Consistency = gocql.ParseConsistency(*cassandraConsistency)
-	cluster.Timeout = time.Duration(*cassandraTimeout) * time.Millisecond
+	cluster.Timeout = cassandra.ConvertTimeout(*cassandraTimeout)
 	cluster.NumConns = *cassandraConcurrency
 	cluster.ProtoVersion = *cqlProtocolVersion
 	cluster.Keyspace = *cassandraKeyspace

cmd/mt-update-ttl/main.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/gocql/gocql"
+	"github.com/grafana/metrictank/store/cassandra"
 	hostpool "github.com/hailocab/go-hostpool"
 	"github.com/raintank/dur"
 )
@@ -23,7 +24,7 @@ var (
 	cassandraKeyspace            = flag.String("cassandra-keyspace", "metrictank", "cassandra keyspace to use for storing the metric data table")
 	cassandraConsistency         = flag.String("cassandra-consistency", "one", "write consistency (any|one|two|three|quorum|all|local_quorum|each_quorum|local_one")
 	cassandraHostSelectionPolicy = flag.String("cassandra-host-selection-policy", "tokenaware,hostpool-epsilon-greedy", "")
-	cassandraTimeout             = flag.Duration("cassandra-timeout", time.Second, "cassandra timeout")
+	cassandraTimeout             = flag.String("cassandra-timeout", "1s", "cassandra timeout")
 	cassandraConcurrency         = flag.Int("cassandra-concurrency", 20, "max number of concurrent reads to cassandra.")
 	cassandraRetries             = flag.Int("cassandra-retries", 0, "how many times to retry a query before failing it")
 	cqlProtocolVersion           = flag.Int("cql-protocol-version", 4, "cql protocol version to use")
@@ -97,7 +98,7 @@ func NewCassandraStore() (*gocql.Session, error) {
 		}
 	}
 	cluster.Consistency = gocql.ParseConsistency(*cassandraConsistency)
-	cluster.Timeout = time.Duration(*cassandraTimeout) * time.Millisecond
+	cluster.Timeout = cassandra.ConvertTimeout(*cassandraTimeout)
 	cluster.NumConns = *cassandraConcurrency
 	cluster.ProtoVersion = *cqlProtocolVersion
 	cluster.Keyspace = *cassandraKeyspace

cmd/mt-whisper-importer-writer/main.go

@@ -89,7 +89,7 @@ func main() {
 	globalFlags.StringVar(&storeConfig.Keyspace, "cassandra-keyspace", storeConfig.Keyspace, "cassandra keyspace to use for storing the metric data table")
 	globalFlags.StringVar(&storeConfig.Consistency, "cassandra-consistency", storeConfig.Consistency, "write consistency (any|one|two|three|quorum|all|local_quorum|each_quorum|local_one")
 	globalFlags.StringVar(&storeConfig.HostSelectionPolicy, "cassandra-host-selection-policy", storeConfig.HostSelectionPolicy, "")
-	globalFlags.DurationVar(&storeConfig.Timeout, "cassandra-timeout", storeConfig.Timeout, "cassandra timeout")
+	globalFlags.StringVar(&storeConfig.Timeout, "cassandra-timeout", storeConfig.Timeout, "cassandra timeout")
 	globalFlags.IntVar(&storeConfig.ReadConcurrency, "cassandra-read-concurrency", storeConfig.ReadConcurrency, "max number of concurrent reads to cassandra.")
 	globalFlags.IntVar(&storeConfig.WriteConcurrency, "cassandra-write-concurrency", storeConfig.WriteConcurrency, "max number of concurrent writes to cassandra.")
 	globalFlags.IntVar(&storeConfig.ReadQueueSize, "cassandra-read-queue-size", storeConfig.ReadQueueSize, "max number of outstanding reads before reads will be dropped. This is important if you run queries that result in many reads in parallel.")

docs/tools.md

@@ -298,8 +298,8 @@ Flags:
     	how many times to retry a query before failing it
   -cassandra-ssl
     	enable SSL connection to cassandra
-  -cassandra-timeout duration
-    	cassandra timeout (default 1s)
+  -cassandra-timeout string
+    	cassandra timeout (default "1s")
   -cassandra-username string
     	username for authentication (default "cassandra")
   -cql-protocol-version int
@@ -364,8 +364,8 @@ Flags:
     	File containing the needed schemas in case database needs initializing (default "/etc/metrictank/schema-store-cassandra.toml")
   -cassandra-ssl
     	enable SSL connection to cassandra
-  -cassandra-timeout duration
-    	cassandra timeout (default 1s)
+  -cassandra-timeout string
+    	cassandra timeout (default "1s")
   -cassandra-username string
     	username for authentication (default "cassandra")
   -config string
@@ -431,8 +431,8 @@ Flags:
     	how many times to retry a query before failing it
   -cassandra-ssl
     	enable SSL connection to cassandra
-  -cassandra-timeout duration
-    	cassandra timeout (default 1s)
+  -cassandra-timeout string
+    	cassandra timeout (default "1s")
   -cassandra-username string
     	username for authentication (default "cassandra")
   -cql-protocol-version int
@@ -498,8 +498,8 @@ Flags:
     	how many times to retry a query before failing it
   -cassandra-ssl
     	enable SSL connection to cassandra
-  -cassandra-timeout int
-    	cassandra timeout in milliseconds (default 1000)
+  -cassandra-timeout string
+    	cassandra timeout (default "1s")
   -cassandra-username string
     	username for authentication (default "cassandra")
   -cql-protocol-version int
@@ -610,8 +610,8 @@ global config flags:
     	File containing the needed schemas in case database needs initializing (default "/etc/metrictank/schema-store-cassandra.toml")
   -cassandra-ssl
     	enable SSL connection to cassandra
-  -cassandra-timeout duration
-    	cassandra timeout (default 1s)
+  -cassandra-timeout string
+    	cassandra timeout (default "1s")
   -cassandra-username string
     	username for authentication (default "cassandra")
   -cassandra-window-factor int

store/cassandra/cassandra.go

@@ -109,6 +109,22 @@ func PrepareChunkData(span uint32, data []byte) []byte {
 	return buf.Bytes()
 }
 
+// the timeout value used to be an integer specifying the number of milliseconds
+// we want to convert it to a duration string, but we need to stay backwards-compatible for now
+func ConvertTimeout(timeout string) time.Duration {
+	if timeoutI, err := strconv.Atoi(timeout); err == nil {
+		log.Warn("cassandra_store: specifying the timeout as integer is deprecated, please use a duration value")
+		return time.Duration(timeoutI) * time.Millisecond
+	} else {
+		timeoutD, err := time.ParseDuration(timeout)
+		if err == nil {
+			return timeoutD
+		}
+		log.Warn("cassandra_store: invalid duration value %s, assuming default (1s)", timeout)
+		return time.Second
+	}
+}
+
 func NewCassandraStore(config *StoreConfig, ttls []uint32) (*CassandraStore, error) {
 	stats.NewGauge32("store.cassandra.write_queue.size").Set(config.WriteQueueSize)
 	stats.NewGauge32("store.cassandra.num_writers").Set(config.WriteConcurrency)
@@ -126,8 +142,9 @@ func NewCassandraStore(config *StoreConfig, ttls []uint32) (*CassandraStore, err
 			Password: config.Password,
 		}
 	}
+
+	cluster.Timeout = ConvertTimeout(config.Timeout)
 	cluster.Consistency = gocql.ParseConsistency(config.Consistency)
-	cluster.Timeout = config.Timeout
 	cluster.ConnectTimeout = cluster.Timeout
 	cluster.NumConns = config.WriteConcurrency
 	cluster.ProtoVersion = config.CqlProtocolVersion

store/cassandra/cassandra_test.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"math"
 	"testing"
+	"time"
 )
 
 type testCase struct {
@@ -60,3 +61,17 @@ func TestGetTTLTables(t *testing.T) {
 		}
 	}
 }
+
+func TestBackwardsCompatibleTimeout(t *testing.T) {
+	checkTimeout := func(input string, expected time.Duration) {
+		timeoutD := ConvertTimeout(input)
+		if timeoutD != expected {
+			t.Fatalf("expected time %s but got %s from input %s", expected.String(), timeoutD.String(), input)
+		}
+	}
+
+	checkTimeout("3500", time.Duration(3500)*time.Millisecond)
+	checkTimeout("3500ms", time.Duration(3500)*time.Millisecond)
+	checkTimeout("3.5s", time.Duration(3500)*time.Millisecond)
+	checkTimeout("nonsense", time.Second)
+}

store/cassandra/config.go

@@ -2,7 +2,6 @@ package cassandra
 
 import (
 	"flag"
-	"time"
 
 	"github.com/rakyll/globalconf"
 )
@@ -12,7 +11,7 @@ type StoreConfig struct {
 	Keyspace                 string
 	Consistency              string
 	HostSelectionPolicy      string
-	Timeout                  time.Duration
+	Timeout                  string
 	ReadConcurrency          int
 	WriteConcurrency         int
 	ReadQueueSize            int
@@ -39,7 +38,7 @@ func NewStoreConfig() *StoreConfig {
 		Keyspace:                 "metrictank",
 		Consistency:              "one",
 		HostSelectionPolicy:      "tokenaware,hostpool-epsilon-greedy",
-		Timeout:                  time.Second,
+		Timeout:                  "1s",
 		ReadConcurrency:          20,
 		WriteConcurrency:         10,
 		ReadQueueSize:            200000,
@@ -68,7 +67,7 @@ func ConfigSetup() *flag.FlagSet {
 	cas.StringVar(&CliConfig.Keyspace, "keyspace", CliConfig.Keyspace, "cassandra keyspace to use for storing the metric data table")
 	cas.StringVar(&CliConfig.Consistency, "consistency", CliConfig.Consistency, "write consistency (any|one|two|three|quorum|all|local_quorum|each_quorum|local_one")
 	cas.StringVar(&CliConfig.HostSelectionPolicy, "host-selection-policy", CliConfig.HostSelectionPolicy, "")
-	cas.DurationVar(&CliConfig.Timeout, "timeout", CliConfig.Timeout, "cassandra timeout")
+	cas.StringVar(&CliConfig.Timeout, "timeout", CliConfig.Timeout, "cassandra timeout")
 	cas.IntVar(&CliConfig.ReadConcurrency, "read-concurrency", CliConfig.ReadConcurrency, "max number of concurrent reads to cassandra.")
 	cas.IntVar(&CliConfig.WriteConcurrency, "write-concurrency", CliConfig.WriteConcurrency, "max number of concurrent writes to cassandra.")
 	cas.IntVar(&CliConfig.ReadQueueSize, "read-queue-size", CliConfig.ReadQueueSize, "max number of outstanding reads before reads will be dropped. This is important if you run queries that result in many reads in parallel.")