Change schema version check to pick up package path (#508)

cmd/server/cadence.go

@@ -53,14 +53,7 @@ func startHandler(c *cli.Context) {
 	log.Printf("config=\n%v\n", cfg.String())
 
 	cassCfg := cfg.Cassandra
-	if err := cassandra.CheckCompatibleVersion(
-		cassCfg, cassCfg.Keyspace, "./schema/cadence/versioned",
-	); err != nil {
-		log.Fatalf("Incompatible versions", err)
-	}
-	if err := cassandra.CheckCompatibleVersion(
-		cassCfg, cassCfg.VisibilityKeyspace, "./schema/visibility/versioned",
-	); err != nil {
+	if err := cassandra.VerifyCompatibleVersion(cassCfg); err != nil {
 		log.Fatalf("Incompatible versions", err)
 	}
 	for _, svc := range getServices(c) {

tools/cassandra/version.go

@@ -21,9 +21,12 @@
 package cassandra
 
 import (
+	"errors"
 	"fmt"
 	"io/ioutil"
+	"path"
 	"regexp"
+	"runtime"
 	"strconv"
 	"strings"
 
@@ -124,16 +127,46 @@ func getExpectedVersion(dir string) (string, error) {
 	return result, nil
 }
 
-// CheckCompatibleVersion check the version compatibility
-func CheckCompatibleVersion(cfg config.Cassandra, keyspace string, dirPath string) error {
+// VerifyCompatibleVersion ensures that the installed version of cadence and visibility keyspaces
+// is greater than or equal to the expected version.
+// In most cases, the versions should match. However if after a schema upgrade there is a code
+// rollback, the code version (expected version) would fall lower than the actual version in
+// cassandra.
+func VerifyCompatibleVersion(cfg config.Cassandra) error {
+	_, filename, _, ok := runtime.Caller(0)
+	if !ok {
+		return errors.New("unable to get the current function path")
+	}
+	return verifyCompatibleVersionWithRoot(cfg, filename)
+}
+
+func verifyCompatibleVersionWithRoot(cfg config.Cassandra, rootFile string) error {
+	// Traverse until project root i.e. "cadence" dir to navigate to the schema/ directory
+	projRoot := rootFile
+	for path.Base(projRoot) != "cadence" {
+		projRoot = path.Dir(projRoot) // According to spec, returns "." when it cannot go any further
+		if projRoot == "." {
+			return fmt.Errorf("Unable to get project root from path: %s", rootFile)
+		}
+	}
+	schemaPath := path.Join(projRoot, "schema/cadence/versioned")
+	if err := checkCompatibleVersion(cfg, cfg.Keyspace, schemaPath); err != nil {
+		return err
+	}
+	schemaPath = path.Join(projRoot, "schema/visibility/versioned")
+	return checkCompatibleVersion(cfg, cfg.VisibilityKeyspace, schemaPath)
+}
+
+// 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)
 	if err != nil {
 		return fmt.Errorf("unable to create CQL Client: %v", err.Error())
 	}
 	defer cqlClient.Close()
 	version, err := cqlClient.ReadSchemaVersion()
 	if err != nil {
-		return fmt.Errorf("unable to read cassandra schema version: %v", err.Error())
+		return fmt.Errorf("unable to read cassandra schema version keyspace: %s error: %v", keyspace, err.Error())
 	}
 	expectedVersion, err := getExpectedVersion(dirPath)
 	if err != nil {

tools/cassandra/version_test.go

@@ -25,6 +25,8 @@ import (
 	"io/ioutil"
 	"math/rand"
 	"os"
+	"path"
+	"runtime"
 	"testing"
 	"time"
 
@@ -159,6 +161,35 @@ func (s *VersionTestSuite) expectedVersionTest(expected string, dirs []string, e
 	}
 }
 
+func (s *VersionTestSuite) TestVerifyCompatibleVersion() {
+	keyspace := "cadence_test"
+	visKeyspace := "cadence_visibility_test"
+	_, filename, _, ok := runtime.Caller(0)
+	s.True(ok)
+	root := path.Dir(path.Dir(path.Dir(filename)))
+	cqlFile := path.Join(root, "schema/cadence/schema.cql")
+	visCqlFile := path.Join(root, "schema/visibility/schema.cql")
+
+	defer s.createKeyspace(keyspace)()
+	defer s.createKeyspace(visKeyspace)()
+	RunTool([]string{
+		"./tool", "-k", keyspace, "-q", "setup-schema", "-f", cqlFile, "-version", "10.0", "-o",
+	})
+	RunTool([]string{
+		"./tool", "-k", visKeyspace, "-q", "setup-schema", "-f", visCqlFile, "-version", "10.0", "-o",
+	})
+
+	cfg := config.Cassandra{
+		Hosts:              "127.0.0.1",
+		Port:               defaultCassandraPort,
+		User:               "",
+		Password:           "",
+		Keyspace:           keyspace,
+		VisibilityKeyspace: visKeyspace,
+	}
+	s.NoError(verifyCompatibleVersionWithRoot(cfg, filename))
+}
+
 func (s *VersionTestSuite) TestCheckCompatibleVersion() {
 	flags := []struct {
 		expectedVersion string
@@ -173,24 +204,30 @@ func (s *VersionTestSuite) TestCheckCompatibleVersion() {
 		{"abc", "1.0", "unable to read expected schema version", true},
 	}
 	for _, flag := range flags {
-		s.checkCompatibleVersion(flag.expectedVersion, flag.actualVersion, flag.errStr, flag.expectedFail)
+		s.runCheckCompatibleVersion(flag.expectedVersion, flag.actualVersion, flag.errStr, flag.expectedFail)
 	}
 }
 
-func (s *VersionTestSuite) checkCompatibleVersion(
-	expected string, actual string, errStr string, expectedFail bool,
-) {
+func (s *VersionTestSuite) createKeyspace(keyspace string) func() {
 	client, err := newCQLClient("127.0.0.1", defaultCassandraPort, "", "", "system")
 	s.NoError(err)
-	defer client.Close()
 
-	r := rand.New(rand.NewSource(time.Now().UnixNano()))
-	keyspace := fmt.Sprintf("version_test_%v", r.Int63())
 	err = client.CreateKeyspace(keyspace, 1)
 	if err != nil {
 		log.Fatalf("error creating keyspace, err=%v", err)
 	}
-	defer client.DropKeyspace(keyspace)
+	return func() {
+		s.NoError(client.DropKeyspace(keyspace))
+		client.Close()
+	}
+}
+
+func (s *VersionTestSuite) runCheckCompatibleVersion(
+	expected string, actual string, errStr string, expectedFail bool,
+) {
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	keyspace := fmt.Sprintf("version_test_%v", r.Int63())
+	defer s.createKeyspace(keyspace)()
 
 	dir := "check_version"
 	tmpDir, err := ioutil.TempDir("", dir)
@@ -219,7 +256,7 @@ func (s *VersionTestSuite) checkCompatibleVersion(
 		User:     "",
 		Password: "",
 	}
-	err = CheckCompatibleVersion(cfg, keyspace, subdir)
+	err = checkCompatibleVersion(cfg, keyspace, subdir)
 	if len(errStr) > 0 {
 		s.Error(err)
 		s.Contains(err.Error(), errStr)