Extract CQL templates from code to config files. Closes #888

Author: beorn-

cmd/mt-index-migrate/main.go

@@ -9,22 +9,22 @@ import (
 
 	"github.com/gocql/gocql"
 	"github.com/grafana/metrictank/cluster/partitioner"
-	"github.com/grafana/metrictank/idx/cassandra"
+	"github.com/grafana/metrictank/util"
 	"github.com/raintank/worldping-api/pkg/log"
 	"gopkg.in/raintank/schema.v1"
 )
 
 var (
-	dryRun          = flag.Bool("dry-run", true, "run in dry-run mode. No changes will be made.")
-	logLevel        = flag.Int("log-level", 2, "log level. 0=TRACE|1=DEBUG|2=INFO|3=WARN|4=ERROR|5=CRITICAL|6=FATAL")
-	srcCassAddr     = flag.String("src-cass-addr", "localhost", "Address of cassandra host to migrate from.")
-	dstCassAddr     = flag.String("dst-cass-addr", "localhost", "Address of cassandra host to migrate to.")
-	srcKeyspace     = flag.String("src-keyspace", "raintank", "Cassandra keyspace in use on source.")
-	dstKeyspace     = flag.String("dst-keyspace", "raintank", "Cassandra keyspace in use on destination.")
-	partitionScheme = flag.String("partition-scheme", "byOrg", "method used for partitioning metrics. (byOrg|bySeries)")
-	numPartitions   = flag.Int("num-partitions", 1, "number of partitions in cluster")
-
-	wg sync.WaitGroup
+	dryRun            = flag.Bool("dry-run", true, "run in dry-run mode. No changes will be made.")
+	logLevel          = flag.Int("log-level", 2, "log level. 0=TRACE|1=DEBUG|2=INFO|3=WARN|4=ERROR|5=CRITICAL|6=FATAL")
+	srcCassAddr       = flag.String("src-cass-addr", "localhost", "Address of cassandra host to migrate from.")
+	dstCassAddr       = flag.String("dst-cass-addr", "localhost", "Address of cassandra host to migrate to.")
+	srcKeyspace       = flag.String("src-keyspace", "raintank", "Cassandra keyspace in use on source.")
+	dstKeyspace       = flag.String("dst-keyspace", "raintank", "Cassandra keyspace in use on destination.")
+	partitionScheme   = flag.String("partition-scheme", "byOrg", "method used for partitioning metrics. (byOrg|bySeries)")
+	numPartitions     = flag.Int("num-partitions", 1, "number of partitions in cluster")
+	cassandraTemplate = flag.String("cassandra-idx-cql-template", "/etc/metrictank/idx-cassandra.toml", "Cassandra CQL toml template for keyspace/keytable IDX creation.")
+	wg                sync.WaitGroup
 )
 
 func main() {
@@ -63,8 +63,11 @@ func main() {
 		log.Fatal(4, "failed to create cql session for destination cassandra. %s", err)
 	}
 
+	// read key(space|table) information from templates
+	keytable_cql_template := util.ReadEntry(*cassandraTemplate, "cassandra-idx.create_keytable_cql_template").(string)
+
 	// ensure the dest table exists.
-	err = dstSession.Query(fmt.Sprintf(cassandra.TableSchema, *dstKeyspace)).Exec()
+	err = dstSession.Query(fmt.Sprintf(keytable_cql_template, *dstKeyspace)).Exec()
 	if err != nil {
 		log.Fatal(4, "cassandra-idx failed to initialize cassandra table. %s", err)
 	}

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

@@ -23,6 +23,8 @@ var (
 	cqlProtocolVersion           = flag.Int("cql-protocol-version", 4, "cql protocol version to use")
 	cassandraCreateKeyspace      = flag.Bool("cassandra-create-keyspace", true, "enable the creation of the metrictank keyspace")
 
+	cassandraTemplate = flag.String("cassandra-cql-template", "/etc/metrictank/storage-cassandra.toml", "Cassandra CQL templates for keyspace/table creation")
+
 	cassandraSSL              = flag.Bool("cassandra-ssl", false, "enable SSL connection to cassandra")
 	cassandraCaPath           = flag.String("cassandra-ca-path", "/etc/metrictank/ca.pem", "cassandra CA certificate path when using SSL")
 	cassandraHostVerification = flag.Bool("cassandra-host-verification", true, "host (hostname and server cert) verification when using SSL")
@@ -69,7 +71,7 @@ func main() {
 		panic(fmt.Sprintf("Error creating directory: %s", err))
 	}
 
-	store, err := cassandra.NewCassandraStore(*cassandraAddrs, *cassandraKeyspace, *cassandraConsistency, *cassandraCaPath, *cassandraUsername, *cassandraPassword, *cassandraHostSelectionPolicy, *cassandraTimeout, cassandraReadConcurrency, cassandraReadConcurrency, cassandraReadQueueSize, 0, *cassandraRetries, *cqlProtocolVersion, windowFactor, cassandraOmitReadTimeout, *cassandraSSL, *cassandraAuth, *cassandraHostVerification, *cassandraCreateKeyspace, ttls)
+	store, err := cassandra.NewCassandraStore(*cassandraAddrs, *cassandraKeyspace, *cassandraConsistency, *cassandraCaPath, *cassandraUsername, *cassandraPassword, *cassandraHostSelectionPolicy, *cassandraTimeout, cassandraReadConcurrency, cassandraReadConcurrency, cassandraReadQueueSize, 0, *cassandraRetries, *cqlProtocolVersion, windowFactor, cassandraOmitReadTimeout, *cassandraSSL, *cassandraAuth, *cassandraHostVerification, *cassandraCreateKeyspace, *cassandraTemplate, ttls)
 	if err != nil {
 		panic(fmt.Sprintf("Failed to instantiate cassandra: %s", err))
 	}

cmd/mt-store-cat/main.go

@@ -43,6 +43,8 @@ var (
 	cqlProtocolVersion      = flag.Int("cql-protocol-version", 4, "cql protocol version to use")
 	cassandraCreateKeyspace = flag.Bool("cassandra-create-keyspace", true, "enable the creation of the metrictank keyspace")
 
+	cassandraTemplate = flag.String("cassandra-cql-template", "/etc/metrictank/storage-cassandra.toml", "Cassandra CQL templates for keyspace/table creation")
+
 	cassandraSSL              = flag.Bool("cassandra-ssl", false, "enable SSL connection to cassandra")
 	cassandraCaPath           = flag.String("cassandra-ca-path", "/etc/metrictank/ca.pem", "cassandra CA certificate path when using SSL")
 	cassandraHostVerification = flag.Bool("cassandra-host-verification", true, "host (hostname and server cert) verification when using SSL")
@@ -162,7 +164,7 @@ func main() {
 		}
 	}
 
-	store, err := cassandra.NewCassandraStore(*cassandraAddrs, *cassandraKeyspace, *cassandraConsistency, *cassandraCaPath, *cassandraUsername, *cassandraPassword, *cassandraHostSelectionPolicy, *cassandraTimeout, *cassandraReadConcurrency, *cassandraReadConcurrency, *cassandraReadQueueSize, 0, *cassandraRetries, *cqlProtocolVersion, *windowFactor, *cassandraOmitReadTimeout, *cassandraSSL, *cassandraAuth, *cassandraHostVerification, *cassandraCreateKeyspace, nil)
+	store, err := cassandra.NewCassandraStore(*cassandraAddrs, *cassandraKeyspace, *cassandraConsistency, *cassandraCaPath, *cassandraUsername, *cassandraPassword, *cassandraHostSelectionPolicy, *cassandraTimeout, *cassandraReadConcurrency, *cassandraReadConcurrency, *cassandraReadQueueSize, 0, *cassandraRetries, *cqlProtocolVersion, *windowFactor, *cassandraOmitReadTimeout, *cassandraSSL, *cassandraAuth, *cassandraHostVerification, *cassandraCreateKeyspace, *cassandraTemplate, nil)
 	if err != nil {
 		log.Fatal(4, "failed to initialize cassandra. %s", err)
 	}

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

@@ -92,6 +92,8 @@ var (
 	cassandraUsername = globalFlags.String("cassandra-username", "cassandra", "username for authentication")
 	cassandraPassword = globalFlags.String("cassandra-password", "cassandra", "password for authentication")
 
+	cassandraTemplate = flag.String("cassandra-cql-template", "/etc/metrictank/storage-cassandra.toml", "Cassandra CQL templates for keyspace/table creation")
+
 	gitHash = "(none)"
 )
 
@@ -151,7 +153,7 @@ func main() {
 		log.SetLevel(log.InfoLevel)
 	}
 
-	store, err := cassandraStore.NewCassandraStore(*cassandraAddrs, *cassandraKeyspace, *cassandraConsistency, *cassandraCaPath, *cassandraUsername, *cassandraPassword, *cassandraHostSelectionPolicy, *cassandraTimeout, *cassandraReadConcurrency, *cassandraReadConcurrency, *cassandraReadQueueSize, 0, *cassandraRetries, *cqlProtocolVersion, *windowFactor, 60, *cassandraSSL, *cassandraAuth, *cassandraHostVerification, *cassandraCreateKeyspace, nil)
+	store, err := cassandraStore.NewCassandraStore(*cassandraAddrs, *cassandraKeyspace, *cassandraConsistency, *cassandraCaPath, *cassandraUsername, *cassandraPassword, *cassandraHostSelectionPolicy, *cassandraTimeout, *cassandraReadConcurrency, *cassandraReadConcurrency, *cassandraReadQueueSize, 0, *cassandraRetries, *cqlProtocolVersion, *windowFactor, 60, *cassandraSSL, *cassandraAuth, *cassandraHostVerification, *cassandraCreateKeyspace, *cassandraTemplate, nil)
 	if err != nil {
 		panic(fmt.Sprintf("Failed to initialize cassandra: %q", err))
 	}

docker/docker-chaos/metrictank.ini

@@ -58,6 +58,8 @@ cassandra-retries = 0
 cql-protocol-version = 4
 # enable the creation of the mdata keyspace and tables, only one node needs this
 cassandra-create-keyspace = false
+# cassandra CQL toml template
+cassandra-cql-template = /etc/metrictank/storage-cassandra.toml
 
 # enable SSL connection to cassandra
 cassandra-ssl = false
@@ -351,6 +353,9 @@ username = cassandra
 password = cassandra
 # enable the creation of the index keyspace and tables, only one node needs this
 create-keyspace = false
+# cassandra CQL toml template
+cassandra-idx-cql-template = /etc/metrictank/idx-cassandra.toml
+
 
 ### in-memory only
 [memory-idx]

docker/docker-cluster/metrictank.ini

@@ -58,7 +58,8 @@ cassandra-retries = 0
 cql-protocol-version = 4
 # enable the creation of the mdata keyspace and tables, only one node needs this
 cassandra-create-keyspace = false
-
+# cassandra CQL toml template
+cassandra-cql-template = /etc/metrictank/storage-cassandra.toml
 # enable SSL connection to cassandra
 cassandra-ssl = false
 # cassandra CA certficate path when using SSL
@@ -351,6 +352,8 @@ username = cassandra
 password = cassandra
 # enable the creation of the index keyspace and tables, only one node needs this
 create-keyspace = false
+# cassandra CQL toml template
+cassandra-idx-cql-template = /etc/metrictank/idx-cassandra.toml
 
 ### in-memory only
 [memory-idx]

docker/docker-dev-custom-cfg-kafka/metrictank.ini

@@ -58,6 +58,8 @@ cassandra-retries = 0
 cql-protocol-version = 4
 # enable the creation of the mdata keyspace and tables, only one node needs this
 cassandra-create-keyspace = true
+# cassandra CQL toml template
+cassandra-cql-template = /etc/metrictank/storage-cassandra.toml
 
 # enable SSL connection to cassandra
 cassandra-ssl = false
@@ -351,6 +353,8 @@ username = cassandra
 password = cassandra
 # enable the creation of the index keyspace and tables, only one node needs this
 create-keyspace = true
+# cassandra CQL toml template
+cassandra-idx-cql-template = /etc/metrictank/idx-cassandra.toml
 
 ### in-memory only
 [memory-idx]

docker/docker-dev/docker-compose.yml

@@ -12,6 +12,11 @@ services:
       - ../../scripts/config/metrictank-docker.ini:/etc/metrictank/metrictank.ini
       - ../../scripts/config/storage-schemas.conf:/etc/metrictank/storage-schemas.conf
       - ../../scripts/config/storage-aggregation.conf:/etc/metrictank/storage-aggregation.conf
+      - ../../scripts/config/storage-cassandra.toml:/etc/metrictank/storage-cassandra.toml
+      - ../../scripts/config/storage-scylladb.toml:/etc/metrictank/storage-scylladb.toml
+      - ../../scripts/config/idx-cassandra.toml:/etc/metrictank/idx-cassandra.toml
+      - ../../scripts/config/idx-scylladb.toml:/etc/metrictank/idx-scylladb.toml
+
     environment:
       WAIT_HOSTS: cassandra:9042
       WAIT_TIMEOUT: 60
@@ -52,4 +57,4 @@ services:
     ports:
      - "8125:8125/udp"
     volumes:
-     - "../statsdaemon.ini:/etc/statsdaemon.ini"
+     - "../statsdaemon.ini:/etc/statsdaemon.ini"

docs/config.md

@@ -88,6 +88,8 @@ cassandra-retries = 0
 cql-protocol-version = 4
 # enable the creation of the mdata keyspace and tables, only one node needs this
 cassandra-create-keyspace = true
+# cassandra CQL toml template
+cassandra-cql-template = /etc/metrictank/storage-cassandra.toml
 # enable SSL connection to cassandra
 cassandra-ssl = false
 # cassandra CA certficate path when using SSL
@@ -412,6 +414,8 @@ username = cassandra
 password = cassandra
 # enable the creation of the index keyspace and tables, only one node needs this
 create-keyspace = true
+# cassandra CQL toml template
+cassandra-idx-cql-template = /etc/metrictank/idx-cassandra.toml
 ```
 
 ### in-memory only

docs/tools.md

@@ -90,6 +90,8 @@ cass config flags:
     	enable cassandra user authentication
   -ca-path string
     	cassandra CA certficate path when using SSL (default "/etc/metrictank/ca.pem")
+  -cassandra-idx-cql-template string
+    	Cassandra CQL template for IDX keyspace creation (default "/etc/metrictank/idx-cassandra.toml")
   -consistency string
     	write consistency (any|one|two|three|quorum|all|local_quorum|each_quorum|local_one (default "one")
   -create-keyspace
@@ -152,6 +154,8 @@ or for resetting partition information when the number of partitions being used
 
 Flags:
 
+  -cassandra-idx-cql-template string
+    	Cassandra CQL toml template for keyspace/keytable IDX creation. (default "/etc/metrictank/idx-cassandra.toml")
   -dry-run
     	run in dry-run mode. No changes will be made. (default true)
   -dst-cass-addr string
@@ -271,6 +275,8 @@ Flags:
     	cassandra CA certificate path when using SSL (default "/etc/metrictank/ca.pem")
   -cassandra-consistency string
     	write consistency (any|one|two|three|quorum|all|local_quorum|each_quorum|local_one (default "one")
+  -cassandra-cql-template string
+    	Cassandra CQL templates for keyspace/table creation (default "/etc/metrictank/storage-cassandra.toml")
   -cassandra-create-keyspace
     	enable the creation of the metrictank keyspace (default true)
   -cassandra-host-selection-policy string
@@ -327,6 +333,8 @@ Flags:
     	cassandra CA certificate path when using SSL (default "/etc/metrictank/ca.pem")
   -cassandra-consistency string
     	write consistency (any|one|two|three|quorum|all|local_quorum|each_quorum|local_one (default "one")
+  -cassandra-cql-template string
+    	Cassandra CQL templates for keyspace/table creation (default "/etc/metrictank/storage-cassandra.toml")
   -cassandra-create-keyspace
     	enable the creation of the metrictank keyspace (default true)
   -cassandra-host-selection-policy string
@@ -598,6 +606,8 @@ cass config flags:
     	enable cassandra user authentication
   -ca-path string
     	cassandra CA certficate path when using SSL (default "/etc/metrictank/ca.pem")
+  -cassandra-idx-cql-template string
+    	Cassandra CQL template for IDX keyspace creation (default "/etc/metrictank/idx-cassandra.toml")
   -consistency string
     	write consistency (any|one|two|three|quorum|all|local_quorum|each_quorum|local_one (default "one")
   -create-keyspace

idx/cassandra/cassandra.go

@@ -13,27 +13,12 @@ import (
 	"github.com/grafana/metrictank/idx"
 	"github.com/grafana/metrictank/idx/memory"
 	"github.com/grafana/metrictank/stats"
+	"github.com/grafana/metrictank/util"
 	"github.com/raintank/worldping-api/pkg/log"
 	"github.com/rakyll/globalconf"
 	"gopkg.in/raintank/schema.v1"
 )
 
-const KeyspaceSchema = `CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}  AND durable_writes = true`
-const TableSchema = `CREATE TABLE IF NOT EXISTS %s.metric_idx (
-    id text,
-    orgid int,
-    partition int,
-    name text,
-    metric text,
-    interval int,
-    unit text,
-    mtype text,
-    tags set<text>,
-    lastupdate int,
-    PRIMARY KEY (partition, id)
-) WITH compaction = {'class': 'SizeTieredCompactionStrategy'}
-    AND compression = {'sstable_compression': 'org.apache.cassandra.io.compress.LZ4Compressor'}`
-
 var (
 	// metric idx.cassadra.query-insert.ok is how many insert queries for a metric completed successfully (triggered by an add or an update)
 	statQueryInsertOk = stats.NewCounter32("idx.cassandra.query-insert.ok")
@@ -63,26 +48,27 @@ var (
 	statSaveSkipped = stats.NewCounter32("idx.cassandra.save.skipped")
 	errmetrics      = cassandra.NewErrMetrics("idx.cassandra")
 
-	Enabled          bool
-	ssl              bool
-	auth             bool
-	hostverification bool
-	createKeyspace   bool
-	keyspace         string
-	hosts            string
-	capath           string
-	username         string
-	password         string
-	consistency      string
-	timeout          time.Duration
-	numConns         int
-	writeQueueSize   int
-	protoVer         int
-	maxStale         time.Duration
-	pruneInterval    time.Duration
-	updateCassIdx    bool
-	updateInterval   time.Duration
-	updateInterval32 uint32
+	Enabled              bool
+	ssl                  bool
+	auth                 bool
+	hostverification     bool
+	createKeyspace       bool
+	cassandraIdxTemplate string
+	keyspace             string
+	hosts                string
+	capath               string
+	username             string
+	password             string
+	consistency          string
+	timeout              time.Duration
+	numConns             int
+	writeQueueSize       int
+	protoVer             int
+	maxStale             time.Duration
+	pruneInterval        time.Duration
+	updateCassIdx        bool
+	updateInterval       time.Duration
+	updateInterval32     uint32
 )
 
 func ConfigSetup() *flag.FlagSet {
@@ -101,6 +87,7 @@ func ConfigSetup() *flag.FlagSet {
 	casIdx.DurationVar(&pruneInterval, "prune-interval", time.Hour*3, "Interval at which the index should be checked for stale series.")
 	casIdx.IntVar(&protoVer, "protocol-version", 4, "cql protocol version to use")
 	casIdx.BoolVar(&createKeyspace, "create-keyspace", true, "enable the creation of the index keyspace and tables, only one node needs this")
+	casIdx.StringVar(&cassandraIdxTemplate, "cassandra-idx-cql-template", "/etc/metrictank/idx-cassandra.toml", "Cassandra CQL template for IDX keyspace creation")
 
 	casIdx.BoolVar(&ssl, "ssl", false, "enable SSL connection to cassandra")
 	casIdx.StringVar(&capath, "ca-path", "/etc/metrictank/ca.pem", "cassandra CA certficate path when using SSL")
@@ -173,13 +160,17 @@ func (c *CasIdx) InitBare() error {
 		return fmt.Errorf("failed to create cassandra session: %s", err)
 	}
 
+	// read templates
+	keyspace_cql_template := util.ReadEntry(cassandraIdxTemplate, "cassandra-idx.create_keyspace_cql_template").(string)
+	keytable_cql_template := util.ReadEntry(cassandraIdxTemplate, "cassandra-idx.create_keytable_cql_template").(string)
+
 	// create the keyspace or ensure it exists
 	if createKeyspace {
-		err = tmpSession.Query(fmt.Sprintf(KeyspaceSchema, keyspace)).Exec()
+		err = tmpSession.Query(fmt.Sprintf(keyspace_cql_template, keyspace)).Exec()
 		if err != nil {
 			return fmt.Errorf("failed to initialize cassandra keyspace: %s", err)
 		}
-		err = tmpSession.Query(fmt.Sprintf(TableSchema, keyspace)).Exec()
+		err = tmpSession.Query(fmt.Sprintf(keytable_cql_template, keyspace)).Exec()
 		if err != nil {
 			return fmt.Errorf("failed to initialize cassandra table: %s", err)
 		}

metrictank-sample.ini

@@ -61,7 +61,8 @@ cassandra-retries = 0
 cql-protocol-version = 4
 # enable the creation of the mdata keyspace and tables, only one node needs this
 cassandra-create-keyspace = true
-
+# cassandra CQL toml template
+cassandra-cql-template = /etc/metrictank/storage-cassandra.toml
 # enable SSL connection to cassandra
 cassandra-ssl = false
 # cassandra CA certficate path when using SSL
@@ -354,6 +355,8 @@ username = cassandra
 password = cassandra
 # enable the creation of the index keyspace and tables, only one node needs this
 create-keyspace = true
+# cassandra CQL toml template
+cassandra-idx-cql-template = /etc/metrictank/idx-cassandra.toml
 
 ### in-memory only
 [memory-idx]