tooling: first pass at cassandra setup-schema tool (#171)

Author: venkat1109

.travis.yml

@@ -19,18 +19,16 @@ addons:
 
 before_install:
   - pip install --user ccm
-  - pip install --user cqlsh==5.0.3
 
 install:
   - go get -u github.com/Masterminds/glide
   - go get github.com/axw/gocov/gocov
   - go get github.com/mattn/goveralls
   - go get golang.org/x/tools/cmd/cover
-  - ccm create test -v 2.2.8 -n 1 -s
-  - sudo ln -sf /home/travis/.local/bin/cqlsh /usr/local/bin/cqlsh
+  - ccm create test -v 3.9 -n 1 -s
 
 script:
   - make cover_ci
 
 after_success:
-  - export BRANCH=$(if [ "$TRAVIS_PULL_REQUEST" == "false" ]; then echo $TRAVIS_BRANCH; else echo $TRAVIS_PULL_REQUEST_BRANCH; fi)
+  - export BRANCH=$(if [ "$TRAVIS_PULL_REQUEST" == "false" ]; then echo $TRAVIS_BRANCH; else echo $TRAVIS_PULL_REQUEST_BRANCH; fi)

Makefile

@@ -18,12 +18,12 @@ THRIFT_SRCS = idl/github.com/uber/cadence/cadence.thrift \
 PROGS = cadence
 TEST_ARG ?= -race -v -timeout 5m
 BUILD := ./build
+TOOLS_CMD_ROOT=./cmd/tools
 
 export PATH := $(GOPATH)/bin:$(PATH)
 
 THRIFT_GEN=$(GOPATH)/bin/thrift-gen
 
-
 define thriftrule
 THRIFT_GEN_SRC += $(THRIFT_GENDIR)/go/$1/tchan-$1.go
 
@@ -43,23 +43,32 @@ ALL_SRC := $(shell find . -name "*.go" | grep -v -e Godeps -e vendor \
 	-e ".*/_.*" \
 	-e ".*/mocks.*")
 
+# filter out the src files for tools
+TOOLS_SRC := $(shell find ./tools -name "*.go")
+TOOLS_SRC += $(TOOLS_CMD_ROOT)
+
 # all directories with *_test.go files in them
 TEST_DIRS := $(sort $(dir $(filter %_test.go,$(ALL_SRC))))
 
-glide:
+vendor/glide.updated: glide.lock glide.yaml
 	glide install
+	touch vendor/glide.updated
 
 clean_thrift:
 	rm -rf .gen
 
-thriftc: clean_thrift glide $(THRIFT_GEN_SRC)
+thriftc: clean_thrift vendor/glide.updated $(THRIFT_GEN_SRC)
 
-bins: thriftc
-	go build -i -o cadence main.go
+cadence-cassandra-tool: vendor/glide.updated $(TOOLS_SRC)
+	go build -i -o cadence-cassandra-tool cmd/tools/cassandra/main.go
 
-bins_nothrift: glide
+cadence: vendor/glide.updated main.go
 	go build -i -o cadence main.go
 
+bins_nothrift: cadence-cassandra-tool cadence
+
+bins: thriftc bins_nothrift
+
 test: bins
 	@rm -f test
 	@rm -f test.log
@@ -86,3 +95,4 @@ cover_ci: cover_profile
 
 clean:
 	rm -rf cadence
+	rm -rf cadence-cassandra-tool

cmd/tools/cassandra/main.go

@@ -0,0 +1,10 @@
+package main
+
+import (
+	"github.com/uber/cadence/tools/cassandra"
+	"os"
+)
+
+func main() {
+	cassandra.RunTool(os.Args)
+}

common/cassandra_helpers.go

@@ -6,11 +6,11 @@ import (
 
 	"github.com/uber/cadence/common/logging"
 
-	"bytes"
-	"os/exec"
-
 	log "github.com/Sirupsen/logrus"
 	"github.com/gocql/gocql"
+	"github.com/uber/cadence/tools/cassandra"
+	"io/ioutil"
+	"os"
 )
 
 // NewCassandraCluster creates a cassandra cluster given comma separated list of clusterHosts
@@ -58,19 +58,39 @@ func DropCassandraKeyspace(s *gocql.Session, keyspace string) (err error) {
 	return
 }
 
-// LoadCassandraSchema loads the schema from the given .cql file on this keyspace using cqlsh
-func LoadCassandraSchema(cqlshpath string, fileName string, keyspace string) (err error) {
-	// Using cqlsh as I couldn't find a way to execute multiple commands through gocql.Session
-	var out bytes.Buffer
-	var stderr bytes.Buffer
-	cmd := exec.Command(cqlshpath, fmt.Sprintf("--keyspace=%v", keyspace), fmt.Sprintf("--file=%v", fileName))
-	cmd.Stdout = &out
-	cmd.Stderr = &stderr
-	err = cmd.Run()
-
-	// CQLSH doesn't return non-zero for some errors
-	if err != nil || len(stderr.String()) > 0 {
-		err = fmt.Errorf("LoadSchema %v returned %v. STDERR: %v", cmd.Path, err, stderr.String())
+// LoadCassandraSchema loads the schema from the given .cql files on this keyspace
+func LoadCassandraSchema(dir string, fileNames []string, keyspace string) (err error) {
+
+	tmpFile, err := ioutil.TempFile("", "_cadence_")
+	if err != nil {
+		return fmt.Errorf("error creating tmp file:%v", err.Error())
+	}
+	defer os.Remove(tmpFile.Name())
+
+	for _, file := range fileNames {
+		content, err := ioutil.ReadFile(dir + "/" + file)
+		if err != nil {
+			return fmt.Errorf("error reading contents of file %v:%v", file, err.Error())
+		}
+		tmpFile.WriteString(string(content))
+		tmpFile.WriteString("\n")
+	}
+
+	tmpFile.Close()
+
+	config := &cassandra.SetupSchemaConfig{
+		BaseConfig: cassandra.BaseConfig{
+			CassHosts:    "127.0.0.1",
+			CassKeyspace: keyspace,
+		},
+		SchemaFilePath:    tmpFile.Name(),
+		Overwrite:         true,
+		DisableVersioning: true,
+	}
+
+	err = cassandra.SetupSchema(config)
+	if err != nil {
+		err = fmt.Errorf("error loading schema:%v", err.Error())
 	}
 	return
 }

common/persistence/persistenceTestBase.go

@@ -719,8 +719,7 @@ func (s *CassandraTestCluster) setupTestCluster(keySpace string, dropKeySpace bo
 	}
 	s.createCluster(testWorkflowClusterHosts, testDatacenter, gocql.Consistency(1), keySpace)
 	s.createKeyspace(1, dropKeySpace)
-	s.loadSchema("workflow_test.cql", schemaDir)
-	s.loadSchema("visibility_test.cql", schemaDir)
+	s.loadSchema([]string{"workflow_test.cql", "visibility_test.cql"}, schemaDir)
 }
 
 func (s *CassandraTestCluster) tearDownTestCluster() {
@@ -757,15 +756,13 @@ func (s *CassandraTestCluster) dropKeyspace() {
 	}
 }
 
-func (s *CassandraTestCluster) loadSchema(fileName string, schemaDir string) {
-	cqlshDir := "cqlsh"
+func (s *CassandraTestCluster) loadSchema(fileNames []string, schemaDir string) {
 	workflowSchemaDir := "./schema/"
-
 	if schemaDir != "" {
 		workflowSchemaDir = schemaDir + "/schema/"
 	}
 
-	err := common.LoadCassandraSchema(cqlshDir, workflowSchemaDir+fileName, s.keyspace)
+	err := common.LoadCassandraSchema(workflowSchemaDir, fileNames, s.keyspace)
 	if err != nil && !strings.Contains(err.Error(), "AlreadyExists") {
 		log.Fatal(err)
 	}

glide.lock

@@ -22,3 +22,4 @@ import:
 - package: github.com/uber-go/timer
   subpackages:
   - twheel
+- package: github.com/urfave/cli

glide.yaml

@@ -0,0 +1,82 @@
+// Copyright (c) 2017 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package cassandra
+
+import "fmt"
+
+type (
+	// baseConfig is the common config
+	// for all of the tasks that work
+	// with cassandra
+	BaseConfig struct {
+		CassHosts    string
+		CassKeyspace string
+	}
+
+	// UpdateSchemaConfig holds the config
+	// params for executing a UpdateSchemaTask
+	UpdateSchemaConfig struct {
+		BaseConfig
+		TargetVersion int
+		SchemaDir     string
+		IsDryRun      bool
+	}
+
+	// SetupSchemaConfig holds the config
+	// params need by the SetupSchemaTask
+	SetupSchemaConfig struct {
+		BaseConfig
+		SchemaFilePath    string
+		InitialVersion    int
+		Overwrite         bool // overwrite previous data
+		DisableVersioning bool // do not use schema versioning
+	}
+
+	// ConfigError is an error type that
+	// represents a problem with the config
+	ConfigError struct {
+		msg string
+	}
+)
+
+const (
+	cliOptEndpoint          = "endpoint"
+	cliOptKeyspace          = "keyspace"
+	cliOptVersion           = "version"
+	cliOptSchemaFile        = "schema-file"
+	cliOptOverwrite         = "overwrite"
+	cliOptDisableVersioning = "disable-versioning"
+
+	cliFlagEndpoint          = cliOptEndpoint + ", ep"
+	cliFlagKeyspace          = cliOptKeyspace + ", k"
+	cliFlagVersion           = cliOptVersion + ", v"
+	cliFlagSchemaFile        = cliOptSchemaFile + ", f"
+	cliFlagOverwrite         = cliOptOverwrite + ", o"
+	cliFlagDisableVersioning = cliOptDisableVersioning + ", d"
+)
+
+func newConfigError(msg string) error {
+	return &ConfigError{msg: msg}
+}
+
+func (e *ConfigError) Error() string {
+	return fmt.Sprintf("Config Error:%v", e.msg)
+}