refactor(backup): unify stage start/end logging

Author: Michal-Leszczynski

pkg/service/backup/service.go

@@ -810,8 +810,10 @@ func (s *Service) Backup(ctx context.Context, clusterID, taskID, runID uuid.UUID
 			}
 			defer clusterSession.Close()
 
-			w.AwaitSchemaAgreement(ctx, clusterSession)
-
+			if err := w.AwaitSchemaAgreement(ctx, clusterSession); err != nil {
+				w.Logger.Info(ctx, "Couldn't await schema agreement, backup of schema as CQL files will be skipped", "error", err)
+				return nil
+			}
 			if err = w.DumpSchema(ctx, clusterSession); err != nil {
 				w.Logger.Info(ctx, "Couldn't dump schema, backup of schema as CQL files will be skipped", "error", err)
 				w.Schema = nil
@@ -848,7 +850,7 @@ func (s *Service) Backup(ctx context.Context, clusterID, taskID, runID uuid.UUID
 	prevStage := run.Stage
 
 	// Execute stages according to the stage order.
-	execStage := func(stage Stage, f func() error) error {
+	execStage := func(stage Stage, f func() error) (err error) {
 		// In purge only mode skip all stages before purge.
 		if target.PurgeOnly {
 			if stage.Index() < StagePurge.Index() {
@@ -874,6 +876,18 @@ func (s *Service) Backup(ctx context.Context, clusterID, taskID, runID uuid.UUID
 		// Always cleanup stats
 		defer w.cleanup(ctx, hi)
 
+		if desc, ok := stageDescription[stage]; ok {
+			up := strings.ToUpper(desc[:1]) + desc[1:]
+			w.Logger.Info(ctx, up+"...")
+			defer func(start time.Time) {
+				if err != nil {
+					w.Logger.Error(ctx, up+" failed see exact errors above", "duration", timeutc.Since(start))
+				} else {
+					w.Logger.Info(ctx, "Done "+desc, "duration", timeutc.Since(start))
+				}
+			}(timeutc.Now())
+		}
+
 		// Run function
 		return errors.Wrap(f(), strings.ReplaceAll(name, "_", " "))
 	}

pkg/service/backup/stage.go

@@ -22,6 +22,18 @@ const (
 	StageDone         Stage = "DONE"
 )
 
+var stageDescription = map[Stage]string{
+	StageInit:         "initialising",
+	StageAwaitSchema:  "awaiting schema agreement",
+	StageSnapshot:     "taking snapshot",
+	StageIndex:        "indexing snapshot files",
+	StageManifest:     "uploading manifest files",
+	StageSchema:       "uploading cql schema",
+	StageUpload:       "uploading snapshot files",
+	StageMoveManifest: "moving manifest files",
+	StagePurge:        "purging stale snapshots",
+}
+
 // StageOrder listing of all stages in the order of execution.
 func StageOrder() []Stage {
 	return []Stage{

pkg/service/backup/worker_index.go

@@ -14,19 +14,9 @@ import (
 	"github.com/scylladb/go-set/strset"
 	"github.com/scylladb/scylla-manager/v3/pkg/scyllaclient"
 	. "github.com/scylladb/scylla-manager/v3/pkg/service/backup/backupspec"
-	"github.com/scylladb/scylla-manager/v3/pkg/util/timeutc"
 )
 
 func (w *worker) Index(ctx context.Context, hosts []hostInfo, limits []DCLimit) (err error) {
-	w.Logger.Info(ctx, "Indexing snapshot files...")
-	defer func(start time.Time) {
-		if err != nil {
-			w.Logger.Error(ctx, "Indexing snapshot files failed see exact errors above", "duration", timeutc.Since(start))
-		} else {
-			w.Logger.Info(ctx, "Done indexing snapshot files", "duration", timeutc.Since(start))
-		}
-	}(timeutc.Now())
-
 	f := func(h hostInfo) error {
 		w.Logger.Info(ctx, "Indexing snapshot files on host", "host", h.IP)
 

pkg/service/backup/worker_manifest.go

@@ -6,25 +6,14 @@ import (
 	"bytes"
 	"context"
 	"net/http"
-	"time"
 
 	"github.com/pkg/errors"
 	"github.com/scylladb/scylla-manager/v3/pkg/scyllaclient"
 	. "github.com/scylladb/scylla-manager/v3/pkg/service/backup/backupspec"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/parallel"
-	"github.com/scylladb/scylla-manager/v3/pkg/util/timeutc"
 )
 
 func (w *worker) UploadManifest(ctx context.Context, hosts []hostInfo) (stepError error) {
-	w.Logger.Info(ctx, "Uploading manifest files...")
-	defer func(start time.Time) {
-		if stepError != nil {
-			w.Logger.Error(ctx, "Uploading manifest files failed see exact errors above", "duration", timeutc.Since(start))
-		} else {
-			w.Logger.Info(ctx, "Done uploading manifest files", "duration", timeutc.Since(start))
-		}
-	}(timeutc.Now())
-
 	// Limit parallelism level, on huge clusters creating manifest content in
 	// memory for all nodes at the same time can lead to memory issues.
 	const maxParallel = 12
@@ -124,15 +113,6 @@ func (w *worker) uploadHostManifest(ctx context.Context, h hostInfo, m ManifestI
 }
 
 func (w *worker) MoveManifest(ctx context.Context, hosts []hostInfo) (err error) {
-	w.Logger.Info(ctx, "Moving manifest files...")
-	defer func(start time.Time) {
-		if err != nil {
-			w.Logger.Error(ctx, "Moving manifest files failed see exact errors above", "duration", timeutc.Since(start))
-		} else {
-			w.Logger.Info(ctx, "Done moving manifest files", "duration", timeutc.Since(start))
-		}
-	}(timeutc.Now())
-
 	rollbacks := make([]func(context.Context) error, len(hosts))
 
 	f := func(i int) (hostErr error) {

pkg/service/backup/worker_purge.go

@@ -4,25 +4,14 @@ package backup
 
 import (
 	"context"
-	"time"
 
 	"github.com/pkg/errors"
 	"github.com/scylladb/go-log"
 	. "github.com/scylladb/scylla-manager/v3/pkg/service/backup/backupspec"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/parallel"
-	"github.com/scylladb/scylla-manager/v3/pkg/util/timeutc"
 )
 
 func (w *worker) Purge(ctx context.Context, hosts []hostInfo, retentionMap RetentionMap) (err error) {
-	w.Logger.Info(ctx, "Purging stale snapshots...")
-	defer func(start time.Time) {
-		if err != nil {
-			w.Logger.Error(ctx, "Purging stale snapshots failed see exact errors above", "duration", timeutc.Since(start))
-		} else {
-			w.Logger.Info(ctx, "Done purging stale snapshots", "duration", timeutc.Since(start))
-		}
-	}(timeutc.Now())
-
 	// List manifests in all locations
 	manifests, err := listManifestsInAllLocations(ctx, w.Client, hosts, w.ClusterID)
 	if err != nil {

pkg/service/backup/worker_schema.go

@@ -12,21 +12,9 @@ import (
 	. "github.com/scylladb/scylla-manager/v3/pkg/service/backup/backupspec"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/parallel"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/retry"
-	"github.com/scylladb/scylla-manager/v3/pkg/util/timeutc"
 )
 
-func (w *workerTools) AwaitSchemaAgreement(ctx context.Context, clusterSession gocqlx.Session) {
-	w.Logger.Info(ctx, "Awaiting schema agreement...")
-
-	var stepError error
-	defer func(start time.Time) {
-		if stepError != nil {
-			w.Logger.Error(ctx, "Awaiting schema agreement failed see exact errors above", "duration", timeutc.Since(start))
-		} else {
-			w.Logger.Info(ctx, "Done awaiting schema agreement", "duration", timeutc.Since(start))
-		}
-	}(timeutc.Now())
-
+func (w *workerTools) AwaitSchemaAgreement(ctx context.Context, clusterSession gocqlx.Session) error {
 	const (
 		waitMin        = 15 * time.Second // nolint: revive
 		waitMax        = 1 * time.Minute
@@ -52,7 +40,7 @@ func (w *workerTools) AwaitSchemaAgreement(ctx context.Context, clusterSession g
 		localSchemaStmt = "SELECT schema_version FROM system.local WHERE key='local'"
 	)
 
-	stepError = retry.WithNotify(ctx, func() error {
+	return retry.WithNotify(ctx, func() error {
 		var v []string
 		if err := clusterSession.Query(peerSchemasStmt, nil).SelectRelease(&v); err != nil {
 			return retry.Permanent(err)
@@ -84,19 +72,10 @@ func (w *worker) DumpSchema(ctx context.Context, clusterSession gocqlx.Session)
 
 func (w *worker) UploadSchema(ctx context.Context, hosts []hostInfo) (stepError error) {
 	if w.Schema == nil {
+		w.Logger.Info(ctx, "No schema CQL file to upload")
 		return nil
 	}
 
-	w.Logger.Info(ctx, "Uploading schema...")
-
-	defer func(start time.Time) {
-		if stepError != nil {
-			w.Logger.Error(ctx, "Uploading schema failed see exact errors above", "duration", timeutc.Since(start))
-		} else {
-			w.Logger.Info(ctx, "Done uploading schema", "duration", timeutc.Since(start))
-		}
-	}(timeutc.Now())
-
 	// Select single host per location
 	locations := map[string]hostInfo{}
 	for _, hi := range hosts {

pkg/service/backup/worker_snapshot.go

@@ -4,23 +4,12 @@ package backup
 
 import (
 	"context"
-	"time"
 
 	"github.com/pkg/errors"
 	. "github.com/scylladb/scylla-manager/v3/pkg/service/backup/backupspec"
-	"github.com/scylladb/scylla-manager/v3/pkg/util/timeutc"
 )
 
 func (w *worker) Snapshot(ctx context.Context, hosts []hostInfo, limits []DCLimit) (err error) {
-	w.Logger.Info(ctx, "Taking snapshots...")
-	defer func(start time.Time) {
-		if err != nil {
-			w.Logger.Error(ctx, "Taking snapshots failed see exact errors above", "duration", timeutc.Since(start))
-		} else {
-			w.Logger.Info(ctx, "Done taking snapshots", "duration", timeutc.Since(start))
-		}
-	}(timeutc.Now())
-
 	f := func(h hostInfo) error {
 		w.Logger.Info(ctx, "Taking snapshots on host", "host", h.IP)
 		err := w.snapshotHost(ctx, h)

pkg/service/backup/worker_upload.go

@@ -14,15 +14,6 @@ import (
 )
 
 func (w *worker) Upload(ctx context.Context, hosts []hostInfo, limits []DCLimit) (err error) {
-	w.Logger.Info(ctx, "Uploading snapshot files...")
-	defer func(start time.Time) {
-		if err != nil {
-			w.Logger.Error(ctx, "Uploading snapshot files failed see exact errors above", "duration", timeutc.Since(start))
-		} else {
-			w.Logger.Info(ctx, "Done uploading snapshot files", "duration", timeutc.Since(start))
-		}
-	}(timeutc.Now())
-
 	f := func(h hostInfo) error {
 		w.Logger.Info(ctx, "Uploading snapshot files on host", "host", h.IP)
 		err := w.uploadHost(ctx, h)