Merge pull request #1307 from percona/release-2.14.0

Release 2.14.0

Author: sandraromanchenko

.github/workflows/ci.yml

@@ -38,9 +38,15 @@ concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
   cancel-in-progress: true
 
+permissions: read-all
+
 jobs:
   test:
     runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      checks: write
+      actions: write
     timeout-minutes: 180
     strategy:
       fail-fast: false
@@ -130,6 +136,9 @@ jobs:
     if: ${{ always() }}
     needs: test
     runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      actions: read
     steps:
       - uses: actions/checkout@v4
       - name: Set up Go

.github/workflows/codecov.yml

@@ -10,6 +10,8 @@ on:
       - main
       - dev
 
+permissions: read-all
+
 jobs:
   go-test:
     name: runner / go-test

.github/workflows/reviewdog.yml

@@ -1,5 +1,10 @@
 name: reviewdog
 on: [pull_request]
+
+permissions:
+  contents: read
+  pull-requests: write
+
 jobs:
   shellcheck:
     name: runner / shellcheck

.github/workflows/trivy.yml

@@ -4,6 +4,8 @@ on:
     branches: ["main", "dev"]
   pull_request:
     branches: ["main", "dev"]
+permissions: read-all
+
 jobs:
   scan:
     name: Trivy

SECURITY.md

@@ -0,0 +1,16 @@
+# Reporting security vulnerabilities
+
+Please report any vulnerabilities privately by emailing our security team at [security@percona.com](mailto:security@percona.com).
+Do not create public Jira issues for security vulnerabilities or disclose details in publicly accessible trackers.
+
+Once a vulnerability has been reported, it should be reviewed, recorded, assigned, and fixed or addressed as appropriate.
+Based initially on an internal severity assessment (for example, using CVSS or our internal severity ratings; if a CVE is later assigned, its criticality may inform or adjust this assessment and may be reduced internally with certain compensating controls that mitigate the severity level), the following guidelines apply to ensure that issues are resolved or addressed in some concrete manner:
+
+1. Critical – immediately, but no later than 14 days;
+2. High – as soon as possible, no later than 30 days;
+3. Medium – within 60 days; and
+4. Low – make best efforts to patch low-rated vulnerabilities within 90 days.
+
+Should you have a legitimate test case that might include one of the above, please contact [security@percona.com](mailto:security@percona.com) detailing your proposed test, expected outcome, and proposed timelines.
+
+For details, see [Percona Security](https://www.percona.com/security).

cmd/pbm-agent/pitr.go

@@ -57,6 +57,58 @@ func (a *Agent) removePitr() {
 	a.setPitr(nil)
 }
 
+// waitForPITRSlicerStop polls until PITR config is disabled and the OpLock is stale.
+func (a *Agent) waitForPITRSlicerStop(ctx context.Context, rs string, l log.LogEvent) error {
+	ctx, cancel := context.WithTimeout(ctx, 120*time.Second)
+	defer cancel()
+
+	tk := time.NewTicker(pitrHb)
+	defer tk.Stop()
+
+	l.Debug("waiting for PITR config disabled and OpLock released")
+	for {
+		select {
+		case <-ctx.Done():
+			return errors.Wrap(ctx.Err(), "done waiting for oplog slicer to stop")
+		case <-tk.C:
+			// Config is disabled
+			cfg, err := config.GetConfig(ctx, a.leadConn)
+			if err != nil {
+				return errors.Wrap(err, "get config")
+			}
+			if cfg.PITR.Enabled {
+				continue
+			}
+
+			// No active lock
+			locks, err := lock.GetOpLocks(ctx, a.leadConn, &lock.LockHeader{
+				Type:    ctrl.CmdPITR,
+				Replset: rs,
+			})
+			if err != nil {
+				return errors.Wrap(err, "get PITR op locks")
+			}
+
+			ts, err := topo.GetClusterTime(ctx, a.leadConn)
+			if err != nil {
+				return errors.Wrap(err, "get cluster time")
+			}
+
+			active := false
+			for i := range locks {
+				if locks[i].Heartbeat.T+defs.StaleFrameSec >= ts.T {
+					active = true
+					break
+				}
+			}
+			if !active {
+				l.Info("PITR slicer stopped")
+				return nil
+			}
+		}
+	}
+}
+
 func (a *Agent) getPitr() *currentPitr {
 	a.slicerMx.Lock()
 	defer a.slicerMx.Unlock()
@@ -819,6 +871,18 @@ func (a *Agent) pitrActivityMonitor(ctx context.Context) {
 				continue
 			}
 
+			// If any RS reported an error, let the error monitor handle it.
+			// Acting here would set StatusReconfig, masking the error and
+			// preventing proper error handling by the slicers on stop.
+			rsErrors, err := oplog.GetReplSetsWithStatus(ctx, a.leadConn, oplog.StatusError)
+			if err != nil && !errors.Is(err, errors.ErrNotFound) {
+				l.Error("activity check RS errors: %v", err)
+				continue
+			}
+			if len(rsErrors) > 0 {
+				continue
+			}
+
 			ackedAgents, err := oplog.GetAgentsWithACK(ctx, a.leadConn)
 			if err != nil {
 				l.Error("activity get acked agents", err)

cmd/pbm-agent/restore.go

@@ -78,39 +78,24 @@ func (a *Agent) Restore(ctx context.Context, r *ctrl.RestoreCmd, opid ctrl.OPID,
 				l.Error("release lock: %v", err)
 			}
 		}()
+	}
 
+	// disable pitr from leader node
+	if nodeInfo.IsClusterLeader() {
 		err = config.SetConfigVar(ctx, a.leadConn, "pitr.enabled", "false")
 		if err != nil {
 			l.Error("disable oplog slicer: %v", err)
 		} else {
 			l.Info("oplog slicer disabled")
 		}
-		a.removePitr()
 	}
 
-	// stop balancer during the restore
-	if a.brief.Sharded && nodeInfo.IsClusterLeader() {
-		bs, err := topo.GetBalancerStatus(ctx, a.leadConn)
-		if err != nil {
-			l.Error("get balancer status: %v", err)
-			return
-		}
-
-		if bs.IsOn() {
-			err := topo.SetBalancerStatus(ctx, a.leadConn, topo.BalancerModeOff)
-			if err != nil {
-				l.Error("set balancer off: %v", err)
-			}
-
-			l.Debug("waiting for balancer off")
-			bs := topo.WaitForBalancerDisabled(ctx, a.leadConn, time.Second*30, l)
-			if bs.IsDisabled() {
-				l.Debug("balancer is disabled")
-			} else {
-				l.Warning("balancer is not disabled: balancer mode: %s, in balancer round: %t",
-					bs.Mode, bs.InBalancerRound)
-			}
-		}
+	// Cancel the local slicer if running.
+	a.removePitr()
+	// Wait for the slicer to stop on every node.
+	if err := a.waitForPITRSlicerStop(ctx, nodeInfo.SetName, l); err != nil {
+		l.Error("unable to stop PITR slicer: %v", err)
+		return
 	}
 
 	var bcpType defs.BackupType

cmd/pbm/backup.go

@@ -170,9 +170,10 @@ func runBackup(
 		}
 		fmt.Printf("Starting backup %q%s", b.name, pinfo)
 	}
-	startCtx, cancel := context.WithTimeout(ctx, cfg.Backup.Timeouts.StartingStatus())
-	defer cancel()
-	err = waitForBcpStatus(startCtx, conn, b.name, showProgress)
+	err = waitForBcpStatus(ctx, conn, b.name, showProgress)
+	if showProgress {
+		fmt.Println()
+	}
 	if err != nil {
 		return nil, errors.Wrap(err, "wait for backup status")
 	}
@@ -213,7 +214,7 @@ func runBackup(
 		}
 
 		if showProgress {
-			fmt.Printf("\nWaiting for '%s' backup...", b.name)
+			fmt.Printf("Waiting for '%s' backup...", b.name)
 		}
 		s, err := waitBackup(ctx, conn, b.name, defs.StatusDone, showProgress)
 		if s != nil && showProgress {
@@ -272,6 +273,10 @@ func waitBackup(
 			case defs.StatusError:
 				return &bcp.Status, bcp.Error()
 			}
+
+			if err := checkBackupStale(ctx, conn, bcp); err != nil {
+				return &bcp.Status, err
+			}
 		}
 
 		if showProgress {
@@ -280,25 +285,69 @@ func waitBackup(
 	}
 }
 
-func waitForBcpStatus(ctx context.Context, conn connect.Client, bcpName string, showProgress bool) error {
+func checkBackupStale(ctx context.Context, conn connect.Client, bcp *backup.BackupMeta) error {
+	clusterTime, err := topo.GetClusterTime(ctx, conn)
+	if err != nil {
+		return errors.Wrap(err, "read cluster time")
+	}
+	if bcp.Hb.T+defs.StaleFrameSec < clusterTime.T {
+		rs := ""
+		for _, s := range bcp.Replsets {
+			rs += fmt.Sprintf("\n- %s: %v", s.Name, s.Status)
+			if s.Error != "" {
+				rs += ": " + s.Error
+			}
+		}
+		return errors.Errorf(
+			"backup stuck at %q status, last heartbeat: %d%s",
+			bcp.Status, bcp.Hb.T, rs)
+	}
+	return nil
+}
+
+func waitForBcpExists(ctx context.Context, conn connect.Client, bcpName string, showProgress bool) error {
 	tk := time.NewTicker(time.Second)
 	defer tk.Stop()
+	to := time.After(defs.WaitBackupStart)
 
-	var bmeta *backup.BackupMeta
 	for {
 		select {
 		case <-tk.C:
 			if showProgress {
 				fmt.Print(".")
 			}
-			var err error
-			bmeta, err = backup.NewDBManager(conn).GetBackupByName(ctx, bcpName)
+			_, err := backup.NewDBManager(conn).GetBackupByName(ctx, bcpName)
 			if errors.Is(err, errors.ErrNotFound) {
 				continue
 			}
 			if err != nil {
 				return errors.Wrap(err, "get backup metadata")
 			}
+			return nil
+		case <-to:
+			return errors.New("no progress from leader, backup metadata not found")
+		}
+	}
+}
+
+func waitForBcpStatus(ctx context.Context, conn connect.Client, bcpName string, showProgress bool) error {
+	if err := waitForBcpExists(ctx, conn, bcpName, showProgress); err != nil {
+		return err
+	}
+
+	tk := time.NewTicker(time.Second)
+	defer tk.Stop()
+
+	for {
+		select {
+		case <-tk.C:
+			if showProgress {
+				fmt.Print(".")
+			}
+			bmeta, err := backup.NewDBManager(conn).GetBackupByName(ctx, bcpName)
+			if err != nil {
+				return errors.Wrap(err, "get backup metadata")
+			}
 			switch bmeta.Status {
 			case defs.StatusRunning, defs.StatusDumpDone, defs.StatusDone, defs.StatusCancelled:
 				return nil
@@ -315,22 +364,12 @@ func waitForBcpStatus(ctx context.Context, conn connect.Client, bcpName string,
 				}
 				return errors.Errorf("status error on %s", rs)
 			}
-		case <-ctx.Done():
-			if bmeta == nil {
-				return errors.New("no progress from leader, backup metadata not found")
-			}
-			rs := ""
-			for _, s := range bmeta.Replsets {
-				rs += fmt.Sprintf("- Backup on replicaset \"%s\" in state: %v\n", s.Name, s.Status)
-				if s.Error != "" {
-					rs += ": " + s.Error
-				}
-			}
-			if rs == "" {
-				rs = "<no replset has started backup>\n"
-			}
 
-			return errors.New("no confirmation that backup has successfully started. Replsets status:\n" + rs)
+			if err := checkBackupStale(ctx, conn, bmeta); err != nil {
+				return err
+			}
+		case <-ctx.Done():
+			return ctx.Err()
 		}
 	}
 }

cmd/pbm/common.go

@@ -1,9 +1,12 @@
 package main
 
 import (
+	"bufio"
 	"context"
 	"encoding/json"
 	"fmt"
+	"os"
+	"strings"
 	"time"
 
 	"github.com/percona/percona-backup-mongodb/pbm/connect"
@@ -15,6 +18,33 @@ import (
 
 var errWaitTimeout = errors.New("Operation is in progress. Check pbm status and logs")
 
+var errUserCanceled = errors.New("canceled")
+
+func askConfirmation(question string) error {
+	fi, err := os.Stdin.Stat()
+	if err != nil {
+		return errors.Wrap(err, "stat stdin")
+	}
+	if (fi.Mode() & os.ModeCharDevice) == 0 {
+		return errors.New("no tty")
+	}
+
+	fmt.Printf("%s [y/N] ", question)
+
+	scanner := bufio.NewScanner(os.Stdin)
+	scanner.Scan()
+	if err := scanner.Err(); err != nil {
+		return errors.Wrap(err, "read stdin")
+	}
+
+	switch strings.TrimSpace(scanner.Text()) {
+	case "yes", "Yes", "YES", "Y", "y":
+		return nil
+	}
+
+	return errUserCanceled
+}
+
 func sendCmd(ctx context.Context, conn connect.Client, cmd ctrl.Cmd) error {
 	cmd.TS = time.Now().UTC().Unix()
 	_, err := conn.CmdStreamCollection().InsertOne(ctx, cmd)