Schema Tool option to configure request timeout for cql client (#900)

fixes #897
cadence-workflow · Jun 27, 2018 · d3e8f10 · d3e8f10
1 parent f11fff3
commit d3e8f10
Show file tree

Hide file tree

Showing 12 changed files with 34 additions and 17 deletions.
diff --git a/common/util.go b/common/util.go
@@ -123,7 +123,7 @@ func CreateFrontendServiceRetryPolicy() backoff.RetryPolicy {
 	return policy
 }
 
-// CreatePersistanceRetryPolicy creates a retry policy for persistence layer operations
+// CreateMatchingRetryPolicy creates a retry policy for calls to matching service
 func CreateMatchingRetryPolicy() backoff.RetryPolicy {
 	policy := backoff.NewExponentialRetryPolicy(matchingServiceOperationInitialInterval)
 	policy.SetMaximumInterval(matchingServiceOperationMaxInterval)

diff --git a/tools/cassandra/config.go b/tools/cassandra/config.go
@@ -35,6 +35,7 @@ type (
 		CassUser     string
 		CassPassword string
 		CassKeyspace string
+		CassTimeout  int
 	}
 
 	// UpdateSchemaConfig holds the config
@@ -76,6 +77,7 @@ const (
 	cliOptPort              = "port"
 	cliOptUser              = "user"
 	cliOptPassword          = "password"
+	cliOptTimeout           = "timeout"
 	cliOptKeyspace          = "keyspace"
 	cliOptVersion           = "version"
 	cliOptSchemaFile        = "schema-file"
@@ -91,6 +93,7 @@ const (
 	cliFlagPort              = cliOptPort + ", p"
 	cliFlagUser              = cliOptUser + ", u"
 	cliFlagPassword          = cliOptPassword + ", pw"
+	cliFlagTimeout           = cliOptTimeout + ", t"
 	cliFlagKeyspace          = cliOptKeyspace + ", k"
 	cliFlagVersion           = cliOptVersion + ", v"
 	cliFlagSchemaFile        = cliOptSchemaFile + ", f"

diff --git a/tools/cassandra/cqlclient.go b/tools/cassandra/cqlclient.go
@@ -74,7 +74,7 @@ var errGetSchemaVersion = errors.New("Failed to get current schema version from
 
 const (
 	newLineDelim         = '\n'
-	defaultTimeout       = 30 * time.Second
+	defaultTimeout       = 30       // timeout in seconds
 	cqlProtoVersion      = 4        // default CQL protocol version
 	defaultConsistency   = "QUORUM" // schema updates must always be QUORUM
 	defaultCassandraPort = 9042
@@ -107,7 +107,8 @@ const (
 )
 
 // newCQLClient returns a new instance of CQLClient
-func newCQLClient(hostsCsv string, port int, user, password, keyspace string) (CQLClient, error) {
+func newCQLClient(hostsCsv string, port int, user, password, keyspace string, timeoutSeconds int) (CQLClient,
+	error) {
 	hosts := parseHosts(hostsCsv)
 	if len(hosts) == 0 {
 		return nil, errNoHosts
@@ -122,8 +123,9 @@ func newCQLClient(hostsCsv string, port int, user, password, keyspace string) (C
 			Password: password,
 		}
 	}
+	timeout := time.Duration(timeoutSeconds) * time.Second
 	clusterCfg.Keyspace = keyspace
-	clusterCfg.Timeout = defaultTimeout
+	clusterCfg.Timeout = timeout
 	clusterCfg.ProtoVersion = cqlProtoVersion
 	clusterCfg.Consistency = gocql.ParseConsistency(defaultConsistency)
 	cqlClient := new(cqlClient)

diff --git a/tools/cassandra/cqlclient_test.go b/tools/cassandra/cqlclient_test.go
@@ -59,7 +59,7 @@ func (s *CQLClientTestSuite) SetupSuite() {
 	rand := rand.New(rand.NewSource(time.Now().UnixNano()))
 	s.keyspace = fmt.Sprintf("cql_client_test_%v", rand.Int63())
 
-	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", "system")
+	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", "system", defaultTimeout)
 	if err != nil {
 		log.Fatalf("error creating CQLClient, err=%v", err)
 	}
@@ -191,7 +191,7 @@ func (s *CQLClientTestSuite) testCreate(client CQLClient) {
 }
 
 func (s *CQLClientTestSuite) TestCQLClient() {
-	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", s.keyspace)
+	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", s.keyspace, defaultTimeout)
 	s.Nil(err)
 	s.testCreate(client)
 	s.testUpdate(client)

diff --git a/tools/cassandra/handler.go b/tools/cassandra/handler.go
@@ -59,7 +59,8 @@ func createKeyspace(cli *cli.Context) error {
 	if err != nil {
 		return handleErr(err)
 	}
-	client, err := newCQLClient(config.CassHosts, config.CassPort, config.CassUser, config.CassPassword, "system")
+	client, err := newCQLClient(config.CassHosts, config.CassPort, config.CassUser, config.CassPassword, "system",
+		config.CassTimeout)
 	if err != nil {
 		return handleErr(fmt.Errorf("error creating cql client:%v", err))
 	}
@@ -127,6 +128,7 @@ func newSetupSchemaConfig(cli *cli.Context) (*SetupSchemaConfig, error) {
 	config.CassPort = cli.GlobalInt(cliOptPort)
 	config.CassUser = cli.GlobalString(cliOptUser)
 	config.CassPassword = cli.GlobalString(cliOptPassword)
+	config.CassTimeout = cli.GlobalInt(cliOptTimeout)
 	config.CassKeyspace = cli.GlobalString(cliOptKeyspace)
 	config.SchemaFilePath = cli.String(cliOptSchemaFile)
 	config.InitialVersion = cli.String(cliOptVersion)
@@ -171,6 +173,7 @@ func newUpdateSchemaConfig(cli *cli.Context) (*UpdateSchemaConfig, error) {
 	config.CassPort = cli.GlobalInt(cliOptPort)
 	config.CassUser = cli.GlobalString(cliOptUser)
 	config.CassPassword = cli.GlobalString(cliOptPassword)
+	config.CassTimeout = cli.GlobalInt(cliOptTimeout)
 	config.CassKeyspace = cli.GlobalString(cliOptKeyspace)
 	config.SchemaDir = cli.String(cliOptSchemaDir)
 	config.IsDryRun = cli.Bool(cliOptDryrun)
@@ -189,6 +192,7 @@ func newCreateKeyspaceConfig(cli *cli.Context) (*CreateKeyspaceConfig, error) {
 	config.CassPort = cli.GlobalInt(cliOptPort)
 	config.CassUser = cli.GlobalString(cliOptUser)
 	config.CassPassword = cli.GlobalString(cliOptPassword)
+	config.CassTimeout = cli.GlobalInt(cliOptTimeout)
 	config.CassKeyspace = cli.String(cliOptKeyspace)
 	config.ReplicationFactor = cli.Int(cliOptReplicationFactor)
 

diff --git a/tools/cassandra/main.go b/tools/cassandra/main.go
@@ -80,6 +80,12 @@ func buildCLIOptions() *cli.App {
 			Usage:  "password used for authentication for connecting to cassandra host",
 			EnvVar: "CASSANDRA_PASSWORD",
 		},
+		cli.IntFlag{
+			Name:   cliFlagTimeout,
+			Value:  defaultTimeout,
+			Usage:  "request timeout in seconds used for cql client",
+			EnvVar: "CASSANDRA_TIMEOUT",
+		},
 		cli.StringFlag{
 			Name:   cliFlagKeyspace,
 			Value:  "cadence",

diff --git a/tools/cassandra/setupTask.go b/tools/cassandra/setupTask.go
@@ -34,7 +34,7 @@ type SetupSchemaTask struct {
 
 func newSetupSchemaTask(config *SetupSchemaConfig) (*SetupSchemaTask, error) {
 	client, err := newCQLClient(config.CassHosts, config.CassPort, config.CassUser, config.CassPassword,
-		config.CassKeyspace)
+		config.CassKeyspace, config.CassTimeout)
 	if err != nil {
 		return nil, err
 	}

diff --git a/tools/cassandra/setupTask_test.go b/tools/cassandra/setupTask_test.go
@@ -61,7 +61,7 @@ func (s *SetupSchemaTestSuite) SetupSuite() {
 	s.rand = rand.New(rand.NewSource(time.Now().UnixNano()))
 	s.keyspace = fmt.Sprintf("setup_schema_test_%v", s.rand.Int63())
 
-	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", "system")
+	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", "system", defaultTimeout)
 	if err != nil {
 		s.log.Fatal("Error creating CQLClient")
 	}
@@ -87,7 +87,7 @@ func (s *SetupSchemaTestSuite) TestCreateKeyspace() {
 
 func (s *SetupSchemaTestSuite) TestSetupSchema() {
 
-	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", s.keyspace)
+	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", s.keyspace, defaultTimeout)
 	s.Nil(err)
 
 	// test command fails without required arguments

diff --git a/tools/cassandra/updateTask.go b/tools/cassandra/updateTask.go
@@ -86,7 +86,8 @@ func NewUpdateSchemaTask(config *UpdateSchemaConfig) (*UpdateSchemaTask, error)
 		}
 	}
 
-	client, err := newCQLClient(config.CassHosts, config.CassPort, config.CassUser, config.CassPassword, keyspace)
+	client, err := newCQLClient(config.CassHosts, config.CassPort, config.CassUser, config.CassPassword, keyspace,
+		config.CassTimeout)
 	if err != nil {
 		return nil, err
 	}
@@ -357,7 +358,8 @@ func readSchemaDir(dir string, startVer string, endVer string) ([]string, error)
 // sets up a temporary dryrun keyspace for
 // executing the cassandra schema update
 func setupDryrunKeyspace(config *UpdateSchemaConfig) error {
-	client, err := newCQLClient(config.CassHosts, config.CassPort, config.CassUser, config.CassPassword, systemKeyspace)
+	client, err := newCQLClient(config.CassHosts, config.CassPort, config.CassUser, config.CassPassword, systemKeyspace,
+		config.CassTimeout)
 	if err != nil {
 		return err
 	}

diff --git a/tools/cassandra/updateTask_test.go b/tools/cassandra/updateTask_test.go
@@ -61,7 +61,7 @@ func (s *UpdateSchemaTestSuite) SetupSuite() {
 	s.rand = rand.New(rand.NewSource(time.Now().UnixNano()))
 	s.keyspace = fmt.Sprintf("update_schema_test_%v", s.rand.Int63())
 
-	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", "system")
+	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", "system", defaultTimeout)
 	if err != nil {
 		s.log.Fatal("Error creating CQLClient")
 	}
@@ -81,7 +81,7 @@ func (s *UpdateSchemaTestSuite) TearDownSuite() {
 
 func (s *UpdateSchemaTestSuite) TestUpdateSchema() {
 
-	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", s.keyspace)
+	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", s.keyspace, defaultTimeout)
 	s.Nil(err)
 	defer client.Close()
 
@@ -118,7 +118,7 @@ func (s *UpdateSchemaTestSuite) TestUpdateSchema() {
 
 func (s *UpdateSchemaTestSuite) TestDryrun() {
 
-	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", s.keyspace)
+	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", s.keyspace, defaultTimeout)
 	s.Nil(err)
 	defer client.Close()
 

diff --git a/tools/cassandra/version.go b/tools/cassandra/version.go
@@ -141,7 +141,7 @@ func VerifyCompatibleVersion(cfg config.Cassandra, rootPath string) error {
 
 // checkCompatibleVersion check the version compatibility
 func checkCompatibleVersion(cfg config.Cassandra, keyspace string, dirPath string) error {
-	cqlClient, err := newCQLClient(cfg.Hosts, cfg.Port, cfg.User, cfg.Password, keyspace)
+	cqlClient, err := newCQLClient(cfg.Hosts, cfg.Port, cfg.User, cfg.Password, keyspace, defaultTimeout)
 	if err != nil {
 		return fmt.Errorf("unable to create CQL Client: %v", err.Error())
 	}

diff --git a/tools/cassandra/version_test.go b/tools/cassandra/version_test.go
@@ -209,7 +209,7 @@ func (s *VersionTestSuite) TestCheckCompatibleVersion() {
 }
 
 func (s *VersionTestSuite) createKeyspace(keyspace string) func() {
-	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", "system")
+	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", "system", defaultTimeout)
 	s.NoError(err)
 
 	err = client.CreateKeyspace(keyspace, 1)