fix(service): close forgotten clients and sessions

Fixes #3769

pkg/scyllaclient/provider.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/scylladb/go-log"
+	"github.com/scylladb/scylla-manager/v3/pkg/util/logutil"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/timeutc"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/uuid"
 )
@@ -88,7 +89,7 @@ func (p *CachedProvider) Close() error {
 
 	for clusterID, c := range p.clients {
 		delete(p.clients, clusterID)
-		c.client.Close()
+		logutil.LogOnError(context.Background(), p.logger, c.client.Close, "Couldn't close scylla client")
 	}
 
 	return nil

pkg/service/cluster/service.go

@@ -22,6 +22,7 @@ import (
 	"github.com/scylladb/scylla-manager/v3/pkg/secrets"
 	"github.com/scylladb/scylla-manager/v3/pkg/service"
 	"github.com/scylladb/scylla-manager/v3/pkg/store"
+	"github.com/scylladb/scylla-manager/v3/pkg/util/logutil"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/uuid"
 	"go.uber.org/multierr"
 )
@@ -141,11 +142,7 @@ func (s *Service) createClient(ctx context.Context, clusterID uuid.UUID) (*scyll
 	if err != nil {
 		return nil, err
 	}
-	defer func() {
-		if err := client.Close(); err != nil {
-			s.logger.Error(ctx, "Couldn't close scylla client", "error", err)
-		}
-	}()
+	defer logutil.LogOnError(ctx, s.logger, client.Close, "Couldn't close scylla client")
 
 	for _, host := range config.Hosts {
 		knownHosts, err := s.discoverHosts(scyllaclient.ClientContextWithSelectedHost(ctx, host), client)
@@ -458,7 +455,7 @@ func (s *Service) validateHostsConnectivity(ctx context.Context, c *Cluster) err
 	if err != nil {
 		return errors.Wrap(err, "create client")
 	}
-	defer client.Close()
+	defer logutil.LogOnError(ctx, s.logger, client.Close, "Couldn't close scylla client")
 
 	status, err := client.Status(ctx)
 	if err != nil {
@@ -544,6 +541,7 @@ func (s *Service) ListNodes(ctx context.Context, clusterID uuid.UUID) ([]Node, e
 	if err != nil {
 		return nil, err
 	}
+	defer logutil.LogOnError(ctx, s.logger, client.Close, "Couldn't close scylla client")
 
 	dcs, err := client.Datacenters(ctx)
 	if err != nil {
@@ -578,6 +576,7 @@ func (s *Service) GetSession(ctx context.Context, clusterID uuid.UUID) (session
 	if err != nil {
 		return session, errors.Wrap(err, "get client")
 	}
+	defer logutil.LogOnError(ctx, s.logger, client.Close, "Couldn't close scylla client")
 
 	ni, err := client.AnyNodeInfo(ctx)
 	if err != nil {

pkg/service/repair/service.go

@@ -156,9 +156,11 @@ func (s *Service) GetTarget(ctx context.Context, clusterID uuid.UUID, properties
 	// Sort plan
 	p.SizeSort()
 	p.PrioritySort(NewInternalTablePreference())
+
 	if clusterSession, err := s.clusterSession(ctx, clusterID); err != nil {
 		s.logger.Info(ctx, "No cluster credentials, couldn't ensure repairing base table before its views", "error", err)
 	} else {
+		defer clusterSession.Close()
 		views, err := query.GetAllViews(clusterSession)
 		if err != nil {
 			return t, errors.Wrap(err, "get cluster views")

pkg/service/repair/service_repair_integration_test.go

@@ -76,12 +76,12 @@ func newRepairTestHelper(t *testing.T, session gocqlx.Session, config repair.Con
 	}
 }
 
-func newRepairWithClusterSessionTestHelper(t *testing.T, session gocqlx.Session, clusterSession gocqlx.Session,
+func newRepairWithClusterSessionTestHelper(t *testing.T, session gocqlx.Session,
 	hrt *HackableRoundTripper, c *scyllaclient.Client, config repair.Config) *repairTestHelper {
 	t.Helper()
 
 	logger := log.NewDevelopmentWithLevel(zapcore.InfoLevel)
-	s := newTestServiceWithClusterSession(t, session, clusterSession, c, config, logger)
+	s := newTestServiceWithClusterSession(t, session, c, config, logger)
 
 	return &repairTestHelper{
 		CommonTestHelper: &CommonTestHelper{
@@ -419,7 +419,7 @@ func newTestService(t *testing.T, session gocqlx.Session, client *scyllaclient.C
 	return s
 }
 
-func newTestServiceWithClusterSession(t *testing.T, session gocqlx.Session, clusterSession gocqlx.Session, client *scyllaclient.Client, c repair.Config, logger log.Logger) *repair.Service {
+func newTestServiceWithClusterSession(t *testing.T, session gocqlx.Session, client *scyllaclient.Client, c repair.Config, logger log.Logger) *repair.Service {
 	t.Helper()
 
 	s, err := repair.NewService(
@@ -430,7 +430,7 @@ func newTestServiceWithClusterSession(t *testing.T, session gocqlx.Session, clus
 			return client, nil
 		},
 		func(ctx context.Context, clusterID uuid.UUID) (gocqlx.Session, error) {
-			return clusterSession, nil
+			return CreateSession(t, client), nil
 		},
 		logger.Named("repair"),
 	)
@@ -667,7 +667,7 @@ func TestServiceRepairOrderIntegration(t *testing.T) {
 
 	session := CreateScyllaManagerDBSession(t)
 	clusterSession := CreateSessionAndDropAllKeyspaces(t, c)
-	h := newRepairWithClusterSessionTestHelper(t, session, clusterSession, hrt, c, repair.DefaultConfig())
+	h := newRepairWithClusterSessionTestHelper(t, session, hrt, c, repair.DefaultConfig())
 
 	// Add prefixes ruining lexicographic order
 	const (
@@ -881,7 +881,7 @@ func TestServiceRepairResumeAllRangesIntegration(t *testing.T) {
 	clusterSession := CreateSessionAndDropAllKeyspaces(t, c)
 	cfg := repair.DefaultConfig()
 	cfg.GracefulStopTimeout = time.Second
-	h := newRepairWithClusterSessionTestHelper(t, session, clusterSession, hrt, c, cfg)
+	h := newRepairWithClusterSessionTestHelper(t, session, hrt, c, cfg)
 
 	const (
 		ks1 = "test_repair_1"

pkg/service/restore/service.go

@@ -69,6 +69,7 @@ func (s *Service) Restore(ctx context.Context, clusterID, taskID, runID uuid.UUI
 	if err != nil {
 		return errors.Wrap(err, "create worker")
 	}
+	defer w.clusterSession.Close()
 	w.setRunInfo(taskID, runID)
 
 	if err := w.initTarget(ctx, properties); err != nil {