Parallelized cluster hosts discovering + async config cache update

pkg/cmd/scylla-manager/server.go

@@ -164,7 +164,7 @@ func (s *server) makeServices(ctx context.Context) error {
 func (s *server) onClusterChange(ctx context.Context, c cluster.Change) error {
 	switch c.Type {
 	case cluster.Update:
-		s.configCacheSvc.ForceUpdateCluster(ctx, c.ID)
+		go s.configCacheSvc.ForceUpdateCluster(ctx, c.ID)
 	case cluster.Create:
 		s.configCacheSvc.ForceUpdateCluster(ctx, c.ID)
 		for _, t := range makeAutoHealthCheckTasks(c.ID) {

pkg/scyllaclient/client_scylla.go

@@ -178,6 +178,15 @@ func (c *Client) Datacenters(ctx context.Context) (map[string][]string, error) {
 	return res, errs
 }
 
+// GossiperEndpointLiveGet finds live nodes (according to gossiper).
+func (c *Client) GossiperEndpointLiveGet(ctx context.Context) ([]string, error) {
+	live, err := c.scyllaOps.GossiperEndpointLiveGet(&operations.GossiperEndpointLiveGetParams{Context: ctx})
+	if err != nil {
+		return nil, err
+	}
+	return live.GetPayload(), nil
+}
+
 // HostDatacenter looks up the datacenter that the given host belongs to.
 func (c *Client) HostDatacenter(ctx context.Context, host string) (dc string, err error) {
 	// Try reading from cache

pkg/scyllaclient/retry_integration_test.go

@@ -57,7 +57,7 @@ func TestRetryWithTimeoutIntegration(t *testing.T) {
 		test := table[i]
 
 		t.Run(fmt.Sprintf("block %d nodes", test.block), func(t *testing.T) {
-			if err := testRetry(hosts, test.block, test.timeout); err != nil {
+			if err := testRetry(t, hosts, test.block, test.timeout); err != nil {
 				t.Fatal(err)
 			}
 		})
@@ -72,12 +72,12 @@ func allHosts() ([]string, error) {
 	return client.Hosts(context.Background())
 }
 
-func testRetry(hosts []string, n int, shouldTimeout bool) error {
+func testRetry(t *testing.T, hosts []string, n int, shouldTimeout bool) error {
 	blockedHosts := make([]string, 0, len(hosts))
 
 	block := func(ctx context.Context, hosts []string) error {
 		for _, h := range hosts {
-			err := RunIptablesCommand(h, CmdBlockScyllaREST)
+			err := RunIptablesCommand(t, h, CmdBlockScyllaREST)
 			if err != nil {
 				return err
 			}
@@ -88,7 +88,7 @@ func testRetry(hosts []string, n int, shouldTimeout bool) error {
 
 	unblock := func(ctx context.Context) error {
 		for _, h := range blockedHosts {
-			err := RunIptablesCommand(h, CmdUnblockScyllaREST)
+			err := RunIptablesCommand(t, h, CmdUnblockScyllaREST)
 			if err != nil {
 				return err
 			}

pkg/service/backup/service_backup_integration_test.go

@@ -859,10 +859,10 @@ func TestBackupWithNodesDownIntegration(t *testing.T) {
 	WriteData(t, clusterSession, testKeyspace, 1)
 
 	Print("Given: downed node")
-	if err := RunIptablesCommand(IPFromTestNet("11"), CmdBlockScyllaREST); err != nil {
+	if err := RunIptablesCommand(t, IPFromTestNet("11"), CmdBlockScyllaREST); err != nil {
 		t.Fatal(err)
 	}
-	defer RunIptablesCommand(IPFromTestNet("11"), CmdUnblockScyllaREST)
+	defer RunIptablesCommand(t, IPFromTestNet("11"), CmdUnblockScyllaREST)
 
 	Print("When: get target")
 	target := backup.Target{

pkg/service/cluster/service.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"sort"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/gocql/gocql"
@@ -34,7 +35,10 @@ type ProviderFunc func(ctx context.Context, id uuid.UUID) (*Cluster, error)
 type ChangeType int8
 
 // ErrNoValidKnownHost is thrown when it was not possible to connect to any of the currently known hosts of the cluster.
-var ErrNoValidKnownHost = errors.New("unable to connect to any of cluster's known hosts")
+var (
+	ErrNoValidKnownHost    = errors.New("unable to connect to any of cluster's known hosts")
+	ErrNoLiveHostAvailable = errors.New("no single live host available")
+)
 
 // ChangeType enumeration.
 const (
@@ -156,7 +160,7 @@ func (s *Service) clientConfig(c *Cluster) scyllaclient.Config {
 }
 
 func (s *Service) discoverAndSetClusterHosts(ctx context.Context, c *Cluster) error {
-	knownHosts, err := s.discoverClusterHosts(ctx, c)
+	knownHosts, _, err := s.discoverClusterHosts(ctx, c)
 	if err != nil {
 		if errors.Is(err, ErrNoValidKnownHost) {
 			s.logger.Error(ctx, "There is no single valid known host for the cluster. "+
@@ -171,55 +175,125 @@ func (s *Service) discoverAndSetClusterHosts(ctx context.Context, c *Cluster) er
 	return errors.Wrap(s.setKnownHosts(c, knownHosts), "update known_hosts in SM DB")
 }
 
-func (s *Service) discoverClusterHosts(ctx context.Context, c *Cluster) ([]string, error) {
-	var contactPoints []string
+const (
+	discoverClusterHostsTimeout = 5 * time.Second
+)
+
+func (s *Service) discoverClusterHosts(ctx context.Context, c *Cluster) (knownHosts, liveHosts []string, err error) {
 	if c.Host != "" {
-		contactPoints = append(contactPoints, c.Host) // Go with the designated contact point first
+		knownHosts, liveHosts, err := s.discoverClusterHostUsingCoordinator(ctx, c, discoverClusterHostsTimeout, c.Host)
+		if err != nil {
+			s.logger.Error(ctx, "Couldn't discover hosts using stored coordinator host, proceeding with other known ones",
+				"coordinator-host", c.Host, "error", err)
+		} else {
+			return knownHosts, liveHosts, nil
+		}
 	} else {
 		s.logger.Error(ctx, "Missing --host flag. Using only previously discovered hosts instead", "cluster ID", c.ID)
 	}
-	contactPoints = append(contactPoints, c.KnownHosts...) // In case it failed, try to contact previously discovered hosts
+	if len(c.KnownHosts) < 1 {
+		return nil, nil, ErrNoValidKnownHost
+	}
 
-	for _, cp := range contactPoints {
-		if cp == "" {
-			s.logger.Error(ctx, "Empty contact point", "cluster ID", c.ID, "contact points", contactPoints)
-			continue
-		}
+	wg := sync.WaitGroup{}
+	type hostsTuple struct {
+		live, known []string
+	}
+	result := make(chan hostsTuple, len(c.KnownHosts))
+	discoverContext, discoverCancel := context.WithCancel(ctx)
+	defer discoverCancel()
 
-		config := scyllaclient.DefaultConfigWithTimeout(s.timeoutConfig)
-		if c.Port != 0 {
-			config.Port = strconv.Itoa(c.Port)
-		}
-		config.AuthToken = c.AuthToken
-		config.Hosts = []string{cp}
+	for _, cp := range c.KnownHosts {
+		wg.Add(1)
 
-		client, err := scyllaclient.NewClient(config, s.logger.Named("client"))
-		if err != nil {
-			s.logger.Error(ctx, "Couldn't connect to contact point", "contact point", cp, "error", err)
-			continue
-		}
+		go func(host string) {
+			defer wg.Done()
 
-		knownHosts, err := s.discoverHosts(ctx, client)
-		logutil.LogOnError(ctx, s.logger, client.Close, "Couldn't close scylla client")
-		if err != nil {
-			s.logger.Error(ctx, "Couldn't discover hosts", "host", cp, "error", err)
-			continue
-		}
-		return knownHosts, nil
+			knownHosts, liveHosts, err := s.discoverClusterHostUsingCoordinator(discoverContext, c, discoverClusterHostsTimeout, host)
+			if err != nil {
+				// Only log if the context hasn't been canceled
+				if !errors.Is(discoverContext.Err(), context.Canceled) {
+					s.logger.Error(ctx, "Couldn't discover hosts", "host", host, "error", err)
+				}
+				return
+			}
+			result <- hostsTuple{
+				live:  liveHosts,
+				known: knownHosts,
+			}
+		}(cp)
+	}
+
+	go func() {
+		wg.Wait()
+		close(result)
+	}()
+
+	// Read results until the channel is closed
+	hosts, ok := <-result
+	if ok {
+		return hosts.known, hosts.live, nil
+	}
+
+	// If no valid results, return error
+	return nil, nil, ErrNoValidKnownHost
+}
+
+func (s *Service) discoverClusterHostUsingCoordinator(ctx context.Context, c *Cluster, apiCallTimeout time.Duration,
+	host string,
+) (knownHosts, liveHosts []string, err error) {
+	config := scyllaclient.DefaultConfigWithTimeout(s.timeoutConfig)
+	if c.Port != 0 {
+		config.Port = strconv.Itoa(c.Port)
 	}
+	config.Timeout = apiCallTimeout
+	config.AuthToken = c.AuthToken
+	config.Hosts = []string{host}
 
-	return nil, ErrNoValidKnownHost
+	client, err := scyllaclient.NewClient(config, s.logger.Named("client"))
+	if err != nil {
+		return nil, nil, err
+	}
+	defer logutil.LogOnError(ctx, s.logger, client.Close, "Couldn't close scylla client")
+
+	liveHosts, err = client.GossiperEndpointLiveGet(ctx)
+	if err != nil {
+		return nil, nil, err
+	}
+	knownHosts, err = s.discoverHosts(ctx, client, liveHosts)
+	if err != nil {
+		return nil, nil, err
+	}
+	return knownHosts, liveHosts, nil
 }
 
 // discoverHosts returns a list of all hosts sorted by DC speed. This is
 // an optimisation for Epsilon-Greedy host pool used internally by
 // scyllaclient.Client that makes it use supposedly faster hosts first.
-func (s *Service) discoverHosts(ctx context.Context, client *scyllaclient.Client) (hosts []string, err error) {
+func (s *Service) discoverHosts(ctx context.Context, client *scyllaclient.Client, liveHosts []string) (hosts []string, err error) {
+	if len(liveHosts) == 0 {
+		return nil, ErrNoLiveHostAvailable
+	}
+
 	dcs, err := client.Datacenters(ctx)
 	if err != nil {
 		return nil, err
 	}
-	closest, err := client.ClosestDC(ctx, dcs)
+	// remove dead nodes from the map
+	liveSet := make(map[string]struct{})
+	for _, host := range liveHosts {
+		liveSet[host] = struct{}{}
+	}
+	filteredDCs := make(map[string][]string)
+	for dc, hosts := range dcs {
+		for _, host := range hosts {
+			if _, isLive := liveSet[host]; isLive {
+				filteredDCs[dc] = append(filteredDCs[dc], host)
+			}
+		}
+	}
+
+	closest, err := client.ClosestDC(ctx, filteredDCs)
 	if err != nil {
 		return nil, err
 	}
@@ -397,7 +471,7 @@ func (s *Service) PutCluster(ctx context.Context, c *Cluster) (err error) {
 	}
 
 	// Check hosts connectivity.
-	if err := s.validateHostsConnectivity(ctx, c); err != nil {
+	if err := s.ValidateHostsConnectivity(ctx, c); err != nil {
 		var tip string
 		switch scyllaclient.StatusCodeOf(err) {
 		case 0:
@@ -487,36 +561,34 @@ func (s *Service) PutCluster(ctx context.Context, c *Cluster) (err error) {
 	return s.notifyChangeListener(ctx, changeEvent)
 }
 
-func (s *Service) validateHostsConnectivity(ctx context.Context, c *Cluster) error {
+// ValidateHostsConnectivity validates that scylla manager agent API is available and responding on all live hosts.
+// Hosts are discovered using cluster.host + cluster.knownHosts saved to the manager's database.
+func (s *Service) ValidateHostsConnectivity(ctx context.Context, c *Cluster) error {
 	if err := s.loadKnownHosts(c); err != nil && !errors.Is(err, gocql.ErrNotFound) {
 		return errors.Wrap(err, "load known hosts")
 	}
 
-	knownHosts, err := s.discoverClusterHosts(ctx, c)
+	knownHosts, liveHosts, err := s.discoverClusterHosts(ctx, c)
 	if err != nil {
 		return errors.Wrap(err, "discover cluster hosts")
 	}
 	c.KnownHosts = knownHosts
 
+	if len(liveHosts) == 0 {
+		return util.ErrValidate(errors.New("no live nodes"))
+	}
+
 	config := s.clientConfig(c)
+	config.Hosts = liveHosts
 	client, err := scyllaclient.NewClient(config, s.logger.Named("client"))
 	if err != nil {
 		return err
 	}
 	defer logutil.LogOnError(ctx, s.logger, client.Close, "Couldn't close scylla client")
 
-	status, err := client.Status(ctx)
-	if err != nil {
-		return errors.Wrap(err, "cluster status")
-	}
-	live := status.Live().Hosts()
-	if len(live) == 0 {
-		return util.ErrValidate(errors.New("no live nodes"))
-	}
-
 	var errs error
-	for i, err := range client.CheckHostsConnectivity(ctx, live) {
-		errs = multierr.Append(errs, errors.Wrap(err, live[i]))
+	for i, err := range client.CheckHostsConnectivity(ctx, liveHosts) {
+		errs = multierr.Append(errs, errors.Wrap(err, liveHosts[i]))
 	}
 	if errs != nil {
 		return util.ErrValidate(errors.Wrap(errs, "connectivity check"))