Merge pull request #603 from zimnx/mz/rest-schema-agreement

Use REST API for schema agreement check

go.mod

@@ -19,6 +19,7 @@ require (
 	github.com/go-openapi/swag v0.19.5
 	github.com/go-openapi/validate v0.19.8
 	github.com/gocql/gocql v0.0.0-20200815110948-5378c8f664e9
+	github.com/golang/snappy v0.0.1 // indirect
 	github.com/google/go-cmp v0.5.2
 	github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed
 	github.com/hashicorp/go-version v1.2.1
@@ -30,7 +31,6 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/scylladb/go-log v0.0.4
 	github.com/scylladb/go-set v1.0.2
-	github.com/scylladb/gocqlx/v2 v2.2.0
 	github.com/shurcooL/githubv4 v0.0.0-20201206200315-234843c633fa
 	github.com/shurcooL/graphql v0.0.0-20200928012149-18c5c3165e3a // indirect
 	github.com/spf13/cobra v1.1.1

go.sum

@@ -253,7 +253,6 @@ github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gobuffalo/flect v0.2.0 h1:EWCvMGGxOjsgwlWaP+f4+Hh6yrrte7JeFL2S6b+0hdM=
 github.com/gobuffalo/flect v0.2.0/go.mod h1:W3K3X9ksuZfir8f/LrfVtWmCDQFfayuylOJ7sz/Fj80=
-github.com/gocql/gocql v0.0.0-20200131111108-92af2e088537/go.mod h1:DL0ekTmBSTdlNF25Orwt/JMzqIq3EJ4MVa/J/uK64OY=
 github.com/gocql/gocql v0.0.0-20200815110948-5378c8f664e9 h1:SBOCi413wRa7i5ZET6dmeg8iqpKO/hE+buwIZ7WhNg4=
 github.com/gocql/gocql v0.0.0-20200815110948-5378c8f664e9/go.mod h1:DL0ekTmBSTdlNF25Orwt/JMzqIq3EJ4MVa/J/uK64OY=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
@@ -511,12 +510,8 @@ github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/scylladb/go-log v0.0.4 h1:/bzRRcdQueZGOAxvRmvAiaPutgyde45KG53cfOfVnTY=
 github.com/scylladb/go-log v0.0.4/go.mod h1:77kz90eX2HHKZUL4Os8on7+8t1CmF6oiSqsgX9Hg+30=
-github.com/scylladb/go-reflectx v1.0.1 h1:b917wZM7189pZdlND9PbIJ6NQxfDPfBvUaQ7cjj1iZQ=
-github.com/scylladb/go-reflectx v1.0.1/go.mod h1:rWnOfDIRWBGN0miMLIcoPt/Dhi2doCMZqwMCJ3KupFc=
 github.com/scylladb/go-set v1.0.2 h1:SkvlMCKhP0wyyct6j+0IHJkBkSZL+TDzZ4E7f7BCcRE=
 github.com/scylladb/go-set v1.0.2/go.mod h1:DkpGd78rljTxKAnTDPFqXSGxvETQnJyuSOQwsHycqfs=
-github.com/scylladb/gocqlx/v2 v2.2.0 h1:dJ2+HrmMkY1sSOXI2eVkcvA01RFJX9KDOkKob+E/v+A=
-github.com/scylladb/gocqlx/v2 v2.2.0/go.mod h1:G1Cuw/NcqM0Mnj5zTRihdx7aW3yBk9zxR+XqCV3aCAA=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=
@@ -709,8 +704,6 @@ golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 h1:SQFwaSi55rU7vdNs9Yr0Z324VNlrF+0wMqRXT4St8ck=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20170830134202-bb24a47a89ea/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

pkg/controllers/cluster/actions/upgrade_version.go

@@ -5,15 +5,12 @@ import (
 	"fmt"
 	"reflect"
 	"sort"
-	"strings"
 	"time"
 
 	"github.com/blang/semver"
-	"github.com/gocql/gocql"
 	"github.com/pkg/errors"
 	"github.com/scylladb/go-log"
 	"github.com/scylladb/go-set/strset"
-	"github.com/scylladb/gocqlx/v2"
 	scyllav1 "github.com/scylladb/scylla-operator/pkg/api/scylla/v1"
 	"github.com/scylladb/scylla-operator/pkg/controllers/cluster/resource"
 	"github.com/scylladb/scylla-operator/pkg/controllers/cluster/util"
@@ -39,9 +36,8 @@ const (
 )
 
 type ClusterVersionUpgrade struct {
-	Cluster        *scyllav1.ScyllaCluster
-	ScyllaClient   *scyllaclient.Client
-	ClusterSession CQLSession
+	Cluster      *scyllav1.ScyllaCluster
+	ScyllaClient *scyllaclient.Client
 
 	ipMapping    map[string]string
 	pollInterval time.Duration
@@ -72,15 +68,6 @@ var ScyllaClientForClusterFunc = func(ctx context.Context, cc client.Client, hos
 	return scyllaclient.NewClient(cfg, logger)
 }
 
-type CQLSession interface {
-	AwaitSchemaAgreement(ctx context.Context) error
-}
-
-var NewSessionFunc = func(hosts []string) (CQLSession, error) {
-	cluster := gocql.NewCluster(hosts...)
-	return gocqlx.WrapSession(cluster.CreateSession())
-}
-
 func (a *ClusterVersionUpgrade) nonMaintenanceHosts(ctx context.Context) ([]string, error) {
 	var hosts []string
 	for _, r := range a.Cluster.Spec.Datacenter.Racks {
@@ -171,11 +158,6 @@ func (a *ClusterVersionUpgrade) genericUpgrade(ctx context.Context) error {
 		return errors.Wrap(err, "create scylla client")
 	}
 
-	a.ClusterSession, err = NewSessionFunc(hosts)
-	if err != nil {
-		return errors.Wrap(err, "create scylla session")
-	}
-
 	if err := a.fsm().Transition(ctx); err != nil {
 		return errors.Wrap(err, "upgrade fsm transition")
 	}
@@ -391,13 +373,7 @@ func (a *ClusterVersionUpgrade) beginUpgrade(ctx context.Context) (fsm.Event, er
 
 func (a *ClusterVersionUpgrade) checkSchemaAgreement(ctx context.Context) (fsm.Event, error) {
 	if err := wait.PollImmediate(a.pollInterval, actionTimeout, func() (bool, error) {
-		if err := a.ClusterSession.AwaitSchemaAgreement(ctx); err != nil {
-			if strings.Contains(err.Error(), "cluster schema versions not consistent") {
-				return false, nil
-			}
-			return false, err
-		}
-		return true, nil
+		return a.ScyllaClient.HasSchemaAgreement(ctx)
 	}); err != nil {
 		return ActionFailure, errors.Wrap(err, "await schema agreement")
 	}

pkg/scyllaclient/client.go

@@ -288,6 +288,19 @@ func (c *Client) IsNativeTransportEnabled(ctx context.Context, host string) (boo
 	return resp.Payload, nil
 }
 
+func (c *Client) HasSchemaAgreement(ctx context.Context) (bool, error) {
+	resp, err := c.scyllaOps.StorageProxySchemaVersionsGet(&scyllaOperations.StorageProxySchemaVersionsGetParams{Context: ctx})
+	if err != nil {
+		return false, err
+	}
+	versions := map[string]struct{}{}
+	for _, kv := range resp.Payload {
+		versions[kv.Key] = struct{}{}
+	}
+
+	return len(versions) == 1, nil
+}
+
 func DefaultTransport() *http.Transport {
 	return &http.Transport{
 		Proxy: http.ProxyFromEnvironment,

pkg/test/integration/scylla.go

@@ -73,6 +73,8 @@ func (s *ScyllaFake) handler(w http.ResponseWriter, r *http.Request) {
 		s.drainRequests++
 	case r.URL.Path == "/storage_service/operation_mode" && r.Method == http.MethodGet:
 		fmt.Fprintf(w, "%q", s.operationalMode)
+	case r.URL.Path == "/storage_proxy/schema_versions" && r.Method == http.MethodGet:
+		fmt.Fprintf(w, `[{"key": "e478e0ca-08f6-39ac-bc9d-a50283aaf200", "value": ["1.1.1.1", "2.2.2.2"]}]`)
 	}
 }
 

test/integration/upgrade_version_test.go

@@ -109,7 +109,6 @@ var _ = Describe("Cluster controller", func() {
 			scylla  *scyllav1.ScyllaCluster
 			sstStub *integration.StatefulSetOperatorStub
 
-			originalActionsNewSessionFunc             func(hosts []string) (actions.CQLSession, error)
 			originalActionsScyllaClientForClusterFunc func(ctx context.Context, cc client.Client, hosts []string, logger log.Logger) (*scyllaclient.Client, error)
 		)
 
@@ -132,12 +131,10 @@ var _ = Describe("Cluster controller", func() {
 				Expect(sstStub.CreatePods(ctx, scylla)).To(Succeed())
 			}
 
-			originalActionsNewSessionFunc = actions.NewSessionFunc
 			originalActionsScyllaClientForClusterFunc = actions.ScyllaClientForClusterFunc
 		})
 
 		AfterEach(func() {
-			actions.NewSessionFunc = originalActionsNewSessionFunc
 			actions.ScyllaClientForClusterFunc = originalActionsScyllaClientForClusterFunc
 			Expect(testEnv.Delete(ctx, scylla)).To(Succeed())
 		})
@@ -182,10 +179,6 @@ var _ = Describe("Cluster controller", func() {
 				return http.DefaultClient.Do(req)
 			}))
 
-			actions.NewSessionFunc = func(hosts []string) (actions.CQLSession, error) {
-				return cqlSessionStub{}, nil
-			}
-
 			actions.ScyllaClientForClusterFunc = func(ctx context.Context, cc client.Client, hosts []string, logger log.Logger) (*scyllaclient.Client, error) {
 				cfg := scyllaclient.DefaultConfig(scyllaAddr)
 				cfg.Transport = hrt
@@ -332,11 +325,3 @@ func markPodReady(pods *corev1.PodList, idx int) error {
 
 	return nil
 }
-
-type cqlSessionStub struct {
-}
-
-func (c cqlSessionStub) AwaitSchemaAgreement(ctx context.Context) error {
-	// Always succeed
-	return nil
-}