fix: various issues related to the automated snapshot management and sync instance (#152)

agneum · agneum · commit f75aecb446b9 · 2020-08-17T15:34:42.000Z
diff --git a/pkg/retrieval/engine/postgres/initialize/physical/physical.go b/pkg/retrieval/engine/postgres/initialize/physical/physical.go
@@ -128,13 +128,23 @@ func (r *RestoreJob) Name() string {
 func (r *RestoreJob) Run(ctx context.Context) (err error) {
 	log.Msg(fmt.Sprintf("Run job: %s. Options: %v", r.Name(), r.CopyOptions))
 
+	defer func() {
+		if err != nil && r.CopyOptions.SyncInstance {
+			if syncErr := r.runSyncInstance(ctx); syncErr != nil {
+				log.Err("Failed to run sync instance", syncErr)
+			}
+		}
+	}()
+
 	isEmpty, err := tools.IsEmptyDirectory(r.globalCfg.DataDir)
 	if err != nil {
 		return errors.Wrap(err, "failed to explore the data directory")
 	}
 
 	if !isEmpty {
-		return errors.New("the data directory is not empty. Clean the data directory before proceeding")
+		log.Msg("Data directory is not empty. Skipping physical restore.")
+
+		return nil
 	}
 
 	if err := tools.PullImage(ctx, r.dockerClient, r.CopyOptions.DockerImage); err != nil {
@@ -213,7 +223,7 @@ func (r *RestoreJob) Run(ctx context.Context) (err error) {
 
 	log.Msg("Running refresh command")
 
-	attachResponse, err := r.dockerClient.ContainerExecAttach(ctx, startCommand.ID, types.ExecStartCheck{Tty: true})
+	attachResponse, err := r.dockerClient.ContainerExecAttach(ctx, startCommand.ID, types.ExecStartCheck{})
 	if err != nil {
 		return errors.Wrap(err, "failed to attach to the exec command")
 	}
@@ -226,12 +236,6 @@ func (r *RestoreJob) Run(ctx context.Context) (err error) {
 
 	log.Msg("Refresh command has been finished")
 
-	if r.CopyOptions.SyncInstance {
-		if err := r.runSyncInstance(ctx); err != nil {
-			log.Err("Failed to run sync instance", err)
-		}
-	}
-
 	return nil
 }
 
@@ -263,7 +267,6 @@ func startingPostgresConfig(pgDataDir string) types.ExecConfig {
 	return types.ExecConfig{
 		AttachStdout: true,
 		AttachStderr: true,
-		Tty:          true,
 		Cmd:          []string{"postgres", "-D", pgDataDir},
 		User:         defaults.Username,
 	}
diff --git a/pkg/retrieval/engine/postgres/initialize/snapshot/physical.go b/pkg/retrieval/engine/postgres/initialize/snapshot/physical.go
@@ -111,21 +111,21 @@ func NewPhysicalInitialJob(cfg config.JobConfig, docker *client.Client, cloneMan
 }
 
 func (p *PhysicalInitial) setupScheduler() error {
-	if p.options.Scheduler == nil {
+	if p.options.Scheduler == nil ||
+		p.options.Scheduler.Snapshot.Timetable == "" && p.options.Scheduler.Retention.Timetable == "" {
 		return nil
 	}
 
 	specParser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
 
-	if _, err := specParser.Parse(p.options.Scheduler.Snapshot.Timetable); err != nil {
+	if _, err := specParser.Parse(p.options.Scheduler.Snapshot.Timetable); p.options.Scheduler.Snapshot.Timetable != "" && err != nil {
 		return errors.Wrapf(err, "failed to parse schedule timetable %q", p.options.Scheduler.Snapshot.Timetable)
 	}
 
-	if _, err := specParser.Parse(p.options.Scheduler.Retention.Timetable); err != nil {
+	if _, err := specParser.Parse(p.options.Scheduler.Retention.Timetable); p.options.Scheduler.Retention.Timetable != "" && err != nil {
 		return errors.Wrapf(err, "failed to parse retention timetable %q", p.options.Scheduler.Retention.Timetable)
 	}
 
-	p.scheduleOnce = sync.Once{}
 	p.scheduler = cron.New()
 
 	return nil
@@ -141,7 +141,7 @@ func (p *PhysicalInitial) Name() string {
 }
 
 // Run starts the job.
-func (p *PhysicalInitial) Run(ctx context.Context) error {
+func (p *PhysicalInitial) Run(ctx context.Context) (err error) {
 	p.scheduleOnce.Do(p.startScheduler(ctx))
 
 	select {
@@ -184,36 +184,42 @@ func (p *PhysicalInitial) Run(ctx context.Context) error {
 		}()
 	}
 
-	// Promotion.
-	if p.options.Promote {
-		// Prepare pre-snapshot.
-		snapshotName, err := p.cloneManager.CreateSnapshot("", preDataStateAt+pre)
-		if err != nil {
-			return errors.Wrap(err, "failed to create a snapshot")
+	defer func() {
+		if _, ok := err.(*skipSnapshotErr); ok {
+			log.Msg(err.Error())
+			err = nil
 		}
+	}()
 
-		defer func() {
+	// Prepare pre-snapshot.
+	snapshotName, err := p.cloneManager.CreateSnapshot("", preDataStateAt+pre)
+	if err != nil {
+		return errors.Wrap(err, "failed to create snapshot")
+	}
+
+	defer func() {
+		if err != nil {
 			if errDestroy := p.cloneManager.DestroySnapshot(snapshotName); errDestroy != nil {
 				log.Err(fmt.Sprintf("Failed to destroy the %q snapshot: %v", snapshotName, err))
 			}
-		}()
-
-		if err := p.cloneManager.CreateClone(cloneName, snapshotName); err != nil {
-			return errors.Wrapf(err, "failed to create a pre clone %s", cloneName)
 		}
+	}()
 
-		defer func() {
+	if err := p.cloneManager.CreateClone(cloneName, snapshotName); err != nil {
+		return errors.Wrapf(err, "failed to create \"pre\" clone %s", cloneName)
+	}
+
+	defer func() {
+		if err != nil {
 			if errDestroy := p.cloneManager.DestroyClone(cloneName); errDestroy != nil {
 				log.Err(fmt.Sprintf("Failed to destroy clone %q: %v", cloneName, err))
 			}
-		}()
+		}
+	}()
 
+	// Promotion.
+	if p.options.Promote {
 		if err := p.promoteInstance(ctx, path.Join(p.globalCfg.MountDir, cloneName)); err != nil {
-			if _, ok := err.(*skipSnapshotErr); ok {
-				log.Msg(err.Error())
-				return nil
-			}
-
 			return err
 		}
 	}
@@ -239,20 +245,25 @@ func (p *PhysicalInitial) Run(ctx context.Context) error {
 }
 
 func (p *PhysicalInitial) startScheduler(ctx context.Context) func() {
-	if p.scheduler == nil {
+	if p.scheduler == nil || p.options.Scheduler == nil ||
+		p.options.Scheduler.Snapshot.Timetable == "" && p.options.Scheduler.Retention.Timetable == "" {
 		return func() {}
 	}
 
 	return func() {
-		if _, err := p.scheduler.AddFunc(p.options.Scheduler.Snapshot.Timetable, p.runAutoSnapshot(ctx)); err != nil {
-			log.Err(errors.Wrap(err, "failed to schedule a new snapshot job"))
-			return
+		if p.options.Scheduler.Snapshot.Timetable != "" {
+			if _, err := p.scheduler.AddFunc(p.options.Scheduler.Snapshot.Timetable, p.runAutoSnapshot(ctx)); err != nil {
+				log.Err(errors.Wrap(err, "failed to schedule a new snapshot job"))
+				return
+			}
 		}
 
-		if _, err := p.scheduler.AddFunc(p.options.Scheduler.Retention.Timetable,
-			p.runAutoCleanup(p.options.Scheduler.Retention.Limit)); err != nil {
-			log.Err(errors.Wrap(err, "failed to schedule a new cleanup job"))
-			return
+		if p.options.Scheduler.Retention.Timetable != "" {
+			if _, err := p.scheduler.AddFunc(p.options.Scheduler.Retention.Timetable,
+				p.runAutoCleanup(p.options.Scheduler.Retention.Limit)); err != nil {
+				log.Err(errors.Wrap(err, "failed to schedule a new cleanup job"))
+				return
+			}
 		}
 
 		p.scheduler.Start()
diff --git a/pkg/services/provision/runners/runners.go b/pkg/services/provision/runners/runners.go
@@ -101,10 +101,9 @@ func (r *LocalRunner) Run(command string, options ...bool) (string, error) {
 	logCommand := Hidden
 	if logsEnabled {
 		logCommand = command
+		log.Dbg(fmt.Sprintf(`Run(Local): "%s"`, logCommand))
 	}
 
-	log.Dbg(fmt.Sprintf(`Run(Local): "%s"`, logCommand))
-
 	var out bytes.Buffer
 	var stderr bytes.Buffer
 
@@ -144,13 +143,10 @@ func (r *LocalRunner) Run(command string, options ...bool) (string, error) {
 
 	outFormatted := strings.Trim(out.String(), " \n")
 
-	logOut := Hidden
 	if logsEnabled {
-		logOut = outFormatted
+		log.Dbg(fmt.Sprintf(`Run(Local): output "%s"`, outFormatted))
 	}
 
-	log.Dbg(fmt.Sprintf(`Run(Local): output "%s"`, logOut))
-
 	if stderrStr := stderr.String(); len(stderrStr) > 0 {
 		log.Dbg("Run(Local): stderr", stderr.String())
 	}
diff --git a/pkg/services/provision/thinclones/zfs/zfs.go b/pkg/services/provision/thinclones/zfs/zfs.go
@@ -24,7 +24,6 @@ const (
 	dataStateAtLabel    = "dblab:datastateat"
 	isRoughStateAtLabel = "dblab:isroughdsa"
 	dataStateAtFormat   = "20060102150405"
-	stdErrCleanupTag    = "cleanup_zfs_snapshot"
 )
 
 // ListEntry defines entry of ZFS list command.
@@ -235,7 +234,6 @@ func RollbackSnapshot(r runners.Runner, pool string, snapshot string) error {
 
 // DestroySnapshot destroys the snapshot.
 func DestroySnapshot(r runners.Runner, snapshotName string) error {
-	// TODO(akartasov): Implement the function.
 	cmd := fmt.Sprintf("zfs destroy -R %s", snapshotName)
 
 	if _, err := r.Run(cmd); err != nil {
@@ -248,11 +246,11 @@ func DestroySnapshot(r runners.Runner, snapshotName string) error {
 // CleanupSnapshots destroys old ZFS snapshots considering retention limit.
 func CleanupSnapshots(r runners.Runner, pool string, retentionLimit int) ([]string, error) {
 	cleanupCmd := fmt.Sprintf(
-		"zfs list -t snapshot -r %s -H -o name -s %s -s creation | grep -v clone | head -n -%d "+
-			"| xargs -n1 --no-run-if-empty zfs destroy -R 2>&1 | logger --stderr --tag \"%s\"",
-		pool, dataStateAtLabel, retentionLimit, stdErrCleanupTag)
+		"zfs list -t snapshot -H -o name -s %s -s creation -r %s | grep -v clone | head -n -%d "+
+			"| xargs -n1 --no-run-if-empty zfs destroy -R ",
+		dataStateAtLabel, pool, retentionLimit)
 
-	out, err := r.Run(cleanupCmd, true)
+	out, err := r.Run(cleanupCmd)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to clean up snapshots")
 	}
