291 lines
7.3 KiB
Go
291 lines
7.3 KiB
Go
package rsync
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"os/exec"
|
|
"regexp"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"syscall"
|
|
|
|
"git.casaderoll.de/michael/urbm/internal/model"
|
|
)
|
|
|
|
type Runner struct {
|
|
Binary string
|
|
mu sync.Mutex
|
|
processes map[string]*os.Process
|
|
}
|
|
|
|
type Progress struct {
|
|
Percent float64
|
|
Bytes int64
|
|
Files int64
|
|
CurrentFile string
|
|
}
|
|
|
|
type Summary struct {
|
|
Bytes int64
|
|
Files int64
|
|
}
|
|
|
|
var (
|
|
progressPattern = regexp.MustCompile(`^\s*([0-9,]+)\s+([0-9]+)%`)
|
|
filesPattern = regexp.MustCompile(`^Number of regular files transferred:\s*([0-9,]+)`)
|
|
bytesPattern = regexp.MustCompile(`^Total transferred file size:\s*([0-9,]+) bytes`)
|
|
)
|
|
|
|
func (r *Runner) Run(ctx context.Context, runID string, job model.Job, callback func(Progress)) (Summary, error) {
|
|
args := Arguments(job)
|
|
cmd := exec.CommandContext(ctx, r.Binary, args...)
|
|
cmd.Env = append(os.Environ(), "LC_ALL=C", "LANG=C")
|
|
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
|
|
cmd.Cancel = func() error {
|
|
if cmd.Process == nil {
|
|
return os.ErrProcessDone
|
|
}
|
|
if err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL); err != nil && !errors.Is(err, syscall.ESRCH) {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
stdout, err := cmd.StdoutPipe()
|
|
if err != nil {
|
|
return Summary{}, err
|
|
}
|
|
stderr, err := cmd.StderrPipe()
|
|
if err != nil {
|
|
return Summary{}, err
|
|
}
|
|
if err := cmd.Start(); err != nil {
|
|
if errors.Is(err, exec.ErrNotFound) {
|
|
return Summary{}, fmt.Errorf("environment: rsync binary not found: %w", err)
|
|
}
|
|
return Summary{}, err
|
|
}
|
|
unregister := r.register(runID, cmd.Process)
|
|
defer unregister()
|
|
|
|
var summary Summary
|
|
done := make(chan struct{})
|
|
go func() {
|
|
scanOutput(stdout, func(line string) {
|
|
if value, ok := parseCount(filesPattern, line); ok {
|
|
summary.Files = value
|
|
}
|
|
if value, ok := parseCount(bytesPattern, line); ok {
|
|
summary.Bytes = value
|
|
}
|
|
if callback != nil {
|
|
if progress, ok := parseProgress(line); ok {
|
|
callback(progress)
|
|
} else if strings.HasPrefix(line, "FILE|") {
|
|
callback(Progress{CurrentFile: strings.TrimPrefix(line, "FILE|")})
|
|
}
|
|
}
|
|
})
|
|
close(done)
|
|
}()
|
|
errBytes, _ := io.ReadAll(io.LimitReader(stderr, 1024*1024))
|
|
err = cmd.Wait()
|
|
<-done
|
|
if err != nil {
|
|
if errors.Is(ctx.Err(), context.Canceled) {
|
|
return Summary{}, fmt.Errorf("cancelled: %w", ctx.Err())
|
|
}
|
|
message := strings.TrimSpace(string(errBytes))
|
|
if message == "" {
|
|
message = err.Error()
|
|
}
|
|
return Summary{}, fmt.Errorf("rsync: %s", message)
|
|
}
|
|
return summary, nil
|
|
}
|
|
|
|
// Estimate performs rsync's complete comparison without writing destination data.
|
|
func (r *Runner) Estimate(ctx context.Context, runID string, job model.Job) (Summary, error) {
|
|
job.Rsync.DryRun = true
|
|
args := Arguments(job)
|
|
for index, arg := range args {
|
|
if strings.HasPrefix(arg, "--out-format=") {
|
|
args[index] = "--out-format="
|
|
}
|
|
}
|
|
cmd := exec.CommandContext(ctx, r.Binary, args...)
|
|
cmd.Env = append(os.Environ(), "LC_ALL=C", "LANG=C")
|
|
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
|
|
cmd.Cancel = func() error {
|
|
if cmd.Process == nil {
|
|
return os.ErrProcessDone
|
|
}
|
|
if err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL); err != nil && !errors.Is(err, syscall.ESRCH) {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
stdout, err := cmd.StdoutPipe()
|
|
if err != nil {
|
|
return Summary{}, err
|
|
}
|
|
stderr, err := cmd.StderrPipe()
|
|
if err != nil {
|
|
return Summary{}, err
|
|
}
|
|
if err := cmd.Start(); err != nil {
|
|
if errors.Is(err, exec.ErrNotFound) {
|
|
return Summary{}, fmt.Errorf("environment: rsync binary not found: %w", err)
|
|
}
|
|
return Summary{}, err
|
|
}
|
|
unregister := r.register(runID, cmd.Process)
|
|
defer unregister()
|
|
var summary Summary
|
|
done := make(chan struct{})
|
|
go func() {
|
|
scanOutput(stdout, func(line string) {
|
|
if value, ok := parseCount(filesPattern, line); ok {
|
|
summary.Files = value
|
|
}
|
|
if value, ok := parseCount(bytesPattern, line); ok {
|
|
summary.Bytes = value
|
|
}
|
|
})
|
|
close(done)
|
|
}()
|
|
errBytes, _ := io.ReadAll(io.LimitReader(stderr, 1024*1024))
|
|
err = cmd.Wait()
|
|
<-done
|
|
if err != nil {
|
|
if errors.Is(ctx.Err(), context.Canceled) {
|
|
return Summary{}, fmt.Errorf("cancelled: %w", ctx.Err())
|
|
}
|
|
return Summary{}, fmt.Errorf("rsync preflight: %s", strings.TrimSpace(string(errBytes)))
|
|
}
|
|
return summary, nil
|
|
}
|
|
|
|
func Arguments(job model.Job) []string {
|
|
options := job.Rsync
|
|
args := []string{"--recursive", "--human-readable", "--info=progress2", "--stats", "--partial", "--out-format=FILE|%n"}
|
|
if !options.Overwrite {
|
|
args = append(args, "--ignore-existing")
|
|
}
|
|
if options.Delete {
|
|
args = append(args, "--delete-delay")
|
|
}
|
|
if options.PreservePermissions {
|
|
args = append(args, "--perms")
|
|
}
|
|
if options.PreserveOwner {
|
|
args = append(args, "--owner")
|
|
}
|
|
if options.PreserveGroup {
|
|
args = append(args, "--group")
|
|
}
|
|
if options.PreserveTimes {
|
|
args = append(args, "--times")
|
|
}
|
|
if options.PreserveLinks {
|
|
args = append(args, "--links")
|
|
}
|
|
if options.PreserveACLs {
|
|
args = append(args, "--acls")
|
|
}
|
|
if options.PreserveXattrs {
|
|
args = append(args, "--xattrs")
|
|
}
|
|
if options.Checksum {
|
|
args = append(args, "--checksum")
|
|
}
|
|
if options.DryRun {
|
|
args = append(args, "--dry-run")
|
|
}
|
|
for _, exclude := range job.Excludes {
|
|
args = append(args, "--exclude", exclude)
|
|
}
|
|
args = append(args, "--")
|
|
for _, source := range job.Sources {
|
|
args = append(args, source.Path)
|
|
}
|
|
args = append(args, strings.TrimRight(options.Target, "/")+"/")
|
|
return args
|
|
}
|
|
|
|
func (r *Runner) Pause(runID string) error { return r.signal(runID, syscall.SIGSTOP) }
|
|
func (r *Runner) Resume(runID string) error { return r.signal(runID, syscall.SIGCONT) }
|
|
|
|
func (r *Runner) signal(runID string, signal syscall.Signal) error {
|
|
r.mu.Lock()
|
|
process := r.processes[runID]
|
|
r.mu.Unlock()
|
|
if process == nil {
|
|
return errors.New("rsync process is not active yet")
|
|
}
|
|
if err := syscall.Kill(-process.Pid, signal); err != nil {
|
|
return fmt.Errorf("signal rsync process group: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Runner) register(runID string, process *os.Process) func() {
|
|
r.mu.Lock()
|
|
if r.processes == nil {
|
|
r.processes = make(map[string]*os.Process)
|
|
}
|
|
r.processes[runID] = process
|
|
r.mu.Unlock()
|
|
return func() {
|
|
r.mu.Lock()
|
|
delete(r.processes, runID)
|
|
r.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func scanOutput(reader io.Reader, callback func(string)) {
|
|
scanner := bufio.NewScanner(reader)
|
|
scanner.Split(splitLinesAndCarriageReturns)
|
|
scanner.Buffer(make([]byte, 64*1024), 2*1024*1024)
|
|
for scanner.Scan() {
|
|
if line := strings.TrimSpace(scanner.Text()); line != "" {
|
|
callback(line)
|
|
}
|
|
}
|
|
}
|
|
|
|
func splitLinesAndCarriageReturns(data []byte, atEOF bool) (advance int, token []byte, err error) {
|
|
for i, value := range data {
|
|
if value == '\n' || value == '\r' {
|
|
return i + 1, data[:i], nil
|
|
}
|
|
}
|
|
if atEOF && len(data) > 0 {
|
|
return len(data), data, nil
|
|
}
|
|
return 0, nil, nil
|
|
}
|
|
|
|
func parseProgress(line string) (Progress, bool) {
|
|
match := progressPattern.FindStringSubmatch(line)
|
|
if len(match) != 3 {
|
|
return Progress{}, false
|
|
}
|
|
bytes, _ := strconv.ParseInt(strings.ReplaceAll(match[1], ",", ""), 10, 64)
|
|
percent, _ := strconv.ParseFloat(match[2], 64)
|
|
return Progress{Bytes: bytes, Percent: percent}, true
|
|
}
|
|
|
|
func parseCount(pattern *regexp.Regexp, line string) (int64, bool) {
|
|
match := pattern.FindStringSubmatch(line)
|
|
if len(match) != 2 {
|
|
return 0, false
|
|
}
|
|
value, err := strconv.ParseInt(strings.ReplaceAll(match[1], ",", ""), 10, 64)
|
|
return value, err == nil
|
|
}
|