Merge branch 'main' of git.digital-infinity.de:DigitalInfinity/clonetool
This commit is contained in:
commit
7e73567f2a
41
README.md
41
README.md
@ -55,13 +55,25 @@ Two subcommands:
|
||||
is used at all — the source agent spawns the write-side helper as a
|
||||
plain local subprocess.
|
||||
- **Self-deploy:** if a remote endpoint has no runnable `clonetool` on
|
||||
`PATH` (or wherever `--remote-bin` points), the manager streams *this*
|
||||
binary to `~/.clonetool/bin/clonetool` on that host over the existing
|
||||
SSH connection and uses it — no install, no root. Disable with
|
||||
`--deploy=false`. If the copied binary won't execute there (wrong CPU
|
||||
architecture) the error says so; build one for the remote's arch
|
||||
`PATH` (or wherever `--remote-bin` points), **or the one that's there is
|
||||
a different build than this binary** (see below), the manager streams
|
||||
*this* binary to `~/.clonetool/bin/clonetool` on that host over the
|
||||
existing SSH connection and uses it — no install, no root. Disable with
|
||||
`--deploy=false` (a stale or missing binary is then an error, except a
|
||||
merely-stale `--remote-bin` is used anyway with a warning, since there's
|
||||
no way to correct it). If the copied binary won't execute there (wrong
|
||||
CPU architecture) the error says so; build one for the remote's arch
|
||||
(`CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build`) and put it on `PATH`
|
||||
or pass `--remote-bin`.
|
||||
- **Version/staleness check:** every binary has a build timestamp baked in
|
||||
(see Build below) and `clonetool version` prints it. Before using a
|
||||
remote `clonetool`, the manager runs its `version` there and compares
|
||||
that timestamp against its own — a mismatch (e.g. the remote was
|
||||
self-deployed from an older build of this tool) is treated the same as
|
||||
"not runnable" and triggers the self-deploy above, so the remote binary
|
||||
is kept in sync with whatever you're running locally. Only the build
|
||||
timestamp is compared, not `GOOS`/`GOARCH` — those are expected to
|
||||
differ across a cross-compiled deploy.
|
||||
- **sudo:** if reading or writing an endpoint that is a **block device**
|
||||
fails with a permission error, `--sudo=auto` (the default) transparently
|
||||
restarts that side's agent — and the helper it spawns on the peer —
|
||||
@ -95,6 +107,14 @@ CGO_ENABLED=0 GOOS=darwin GOARCH=arm64 go build -o clonetool-darwin .
|
||||
```
|
||||
|
||||
`./build.sh` writes all three of the above (plus linux/amd64) into `dist/`.
|
||||
It also stamps every binary it builds with the same build timestamp (via
|
||||
`-ldflags -X main.buildTime=...`), which is how the manager tells a stale
|
||||
self-deployed remote binary apart from a current one (see "Version/staleness
|
||||
check" above) — build all your binaries for a given release with one
|
||||
`./build.sh` run so they compare equal to each other. A plain `go build`
|
||||
with no `-ldflags` leaves the timestamp at `"dev"`, which still works for
|
||||
this check (it only ever compares equal to another literal copy of the same
|
||||
`dev` binary, never to a real timestamp).
|
||||
|
||||
Copy the resulting binary to the manager host. Source and destination
|
||||
hosts get it automatically (see self-deploy above), or place it yourself
|
||||
@ -300,8 +320,15 @@ Options:
|
||||
| `--remote-bin` | `clonetool` | Path to clonetool on remote hosts. |
|
||||
| `--manager-host` | local hostname | Address a peer should use to reach this machine, needed only when source or dest is local to the manager *and* the other side is remote and ends up needing to dial back in (pull fallback). |
|
||||
|
||||
`clonetool version` prints the binary's `GOOS/GOARCH` (used internally for
|
||||
the self-deploy check).
|
||||
`clonetool version` prints the binary's `GOOS/GOARCH` and build timestamp,
|
||||
e.g. `clonetool linux/amd64 build=2024-06-01T12:00:00Z` (the timestamp is
|
||||
used internally for the self-deploy staleness check above).
|
||||
|
||||
The status line shows read throughput for both sides and write throughput
|
||||
for the destination separately: `rd(src)` is the source reading and
|
||||
comparing its blocks, `rd(dst)` is the destination fingerprinting its
|
||||
current content (only while the trailing `scan H/M` segment is present),
|
||||
and `wr(dst)` is the destination actually writing changed blocks.
|
||||
|
||||
## Caveats
|
||||
|
||||
|
||||
76
agent.go
76
agent.go
@ -252,7 +252,7 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
var pendingMu sync.Mutex
|
||||
pending := make(map[uint64]bool)
|
||||
|
||||
var copied, skipped int64
|
||||
var copied, skipped, srcRead int64
|
||||
var lastProgress time.Time
|
||||
maybeProgress := func() {
|
||||
if time.Since(lastProgress) < 500*time.Millisecond {
|
||||
@ -260,11 +260,24 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
}
|
||||
lastProgress = time.Now()
|
||||
_ = out.WriteJSON(CtrlMsg{
|
||||
Type: msgProgress, Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped),
|
||||
Type: msgProgress, Phase: "xfer",
|
||||
Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped), SrcRead: atomic.LoadInt64(&srcRead),
|
||||
TotalBlocks: blockCount,
|
||||
})
|
||||
}
|
||||
|
||||
// lastActivityNano records the last moment either side made progress: a
|
||||
// frame arriving from the sink (hashes, ACKs, relayed progress) or the
|
||||
// source loop reading and comparing a block. The idle watchdog below
|
||||
// treats a stall as fatal only when *neither* side has moved for a whole
|
||||
// window — otherwise a check-only run, where every block matches and so no
|
||||
// DATA frames and no ACKs are ever exchanged, trips the watchdog purely
|
||||
// because the wire went quiet while the two scans churned through
|
||||
// identical data.
|
||||
var lastActivityNano int64
|
||||
atomic.StoreInt64(&lastActivityNano, time.Now().UnixNano())
|
||||
markActivity := func() { atomic.StoreInt64(&lastActivityNano, time.Now().UnixNano()) }
|
||||
|
||||
// hashCh is sized to hold every block hash so the reader below never
|
||||
// blocks handing hashes off (which, since ACKs share the same stream,
|
||||
// would otherwise be able to deadlock against the in-flight-send limit).
|
||||
@ -295,6 +308,7 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
}
|
||||
return
|
||||
}
|
||||
markActivity()
|
||||
switch typ {
|
||||
case frameBlockHash:
|
||||
idx, h, derr := decodeBlockHashFrame(payload)
|
||||
@ -335,9 +349,11 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
sendErrCh := make(chan error, 1)
|
||||
go func() {
|
||||
sendErrCh <- runSourceLoop(sourceLoopParams{
|
||||
File: srcFile, Base: req.Base, Size: req.Size, BlockSize: req.BlockSize, Align: align, Hashes: hashCh, Out: fw,
|
||||
OnSkip: func(uint64) { atomic.AddInt64(&skipped, 1); maybeProgress() },
|
||||
File: srcFile, Size: req.Size, BlockSize: req.BlockSize, Align: align, Hashes: hashCh, Out: fw,
|
||||
OnRead: func(uint64) { markActivity(); atomic.AddInt64(&srcRead, 1); maybeProgress() },
|
||||
OnSkip: func(uint64) { markActivity(); atomic.AddInt64(&skipped, 1); maybeProgress() },
|
||||
OnSend: func(idx uint64, _ [32]byte) {
|
||||
markActivity()
|
||||
sem <- struct{}{}
|
||||
pendingMu.Lock()
|
||||
pending[idx] = true
|
||||
@ -353,7 +369,11 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
sendCh := sendErrCh
|
||||
ackCh := ackEvents
|
||||
var fatalErr error
|
||||
idle := time.NewTimer(120 * time.Second)
|
||||
// idleTimeout is how long *both* sides can be silent before the transfer
|
||||
// is declared stalled. The timer may fire sooner; a fire is only fatal
|
||||
// when lastActivityNano confirms nothing has moved for the whole window.
|
||||
const idleTimeout = 120 * time.Second
|
||||
idle := time.NewTimer(idleTimeout)
|
||||
defer idle.Stop()
|
||||
for sendCh != nil || ackCh != nil {
|
||||
if !idle.Stop() {
|
||||
@ -362,7 +382,7 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
default:
|
||||
}
|
||||
}
|
||||
idle.Reset(120 * time.Second)
|
||||
idle.Reset(idleTimeout)
|
||||
select {
|
||||
case sendErr := <-sendCh:
|
||||
sendCh = nil
|
||||
@ -388,7 +408,11 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
maybeProgress()
|
||||
}
|
||||
case <-idle.C:
|
||||
fatalErr = fmt.Errorf("timed out waiting for the sink")
|
||||
idleFor := time.Since(time.Unix(0, atomic.LoadInt64(&lastActivityNano)))
|
||||
if idleFor < idleTimeout {
|
||||
continue // progress somewhere within the window; re-arm and wait
|
||||
}
|
||||
fatalErr = fmt.Errorf("timed out waiting for the sink (no activity for %s)", idleFor.Round(time.Second))
|
||||
}
|
||||
if fatalErr != nil {
|
||||
break
|
||||
@ -405,7 +429,11 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
return fmt.Errorf("sink closed the connection with %d block write confirmation(s) still outstanding", n)
|
||||
}
|
||||
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped), TotalBlocks: blockCount})
|
||||
_ = out.WriteJSON(CtrlMsg{
|
||||
Type: msgProgress, Phase: "xfer",
|
||||
Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped), SrcRead: atomic.LoadInt64(&srcRead),
|
||||
TotalBlocks: blockCount,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -484,6 +512,8 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
|
||||
hashErrCh <- err
|
||||
}()
|
||||
|
||||
var written int64
|
||||
var lastWritten time.Time
|
||||
loopErr := runDestLoop(destLoopParams{
|
||||
File: dstFile, Base: req.Base, BlockSize: req.BlockSize, Align: align, In: fr,
|
||||
OnCtrlMsg: func(m CtrlMsg) {
|
||||
@ -491,7 +521,15 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
|
||||
_ = out.WriteJSON(m)
|
||||
}
|
||||
},
|
||||
OnWrite: func(uint64) {
|
||||
written++
|
||||
if time.Since(lastWritten) >= 500*time.Millisecond {
|
||||
lastWritten = time.Now()
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "write", Written: written})
|
||||
}
|
||||
},
|
||||
})
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "write", Written: written})
|
||||
hashErr := <-hashErrCh
|
||||
|
||||
if hashErr != nil {
|
||||
@ -555,7 +593,19 @@ func runSinkRole(path string, base, size, blockSize int64) error {
|
||||
hashErrCh <- err
|
||||
}()
|
||||
|
||||
loopErr := runDestLoop(destLoopParams{File: f, Base: base, BlockSize: blockSize, Align: align, In: in, AckOut: out})
|
||||
var written int64
|
||||
var lastWritten time.Time
|
||||
loopErr := runDestLoop(destLoopParams{
|
||||
File: f, BlockSize: blockSize, Align: align, In: in, AckOut: out,
|
||||
OnWrite: func(uint64) {
|
||||
written++
|
||||
if time.Since(lastWritten) >= 500*time.Millisecond {
|
||||
lastWritten = time.Now()
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "write", Written: written})
|
||||
}
|
||||
},
|
||||
})
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "write", Written: written})
|
||||
if hashErr := <-hashErrCh; hashErr != nil {
|
||||
return hashErr
|
||||
}
|
||||
@ -621,21 +671,23 @@ func runSourceStreamRole(path string, base, size, blockSize int64) error {
|
||||
}
|
||||
}()
|
||||
|
||||
var copied, skipped int64
|
||||
var copied, skipped, srcRead int64
|
||||
var lastProgress time.Time
|
||||
maybeProgress := func() {
|
||||
if time.Since(lastProgress) < 500*time.Millisecond {
|
||||
return
|
||||
}
|
||||
lastProgress = time.Now()
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Copied: copied, Skipped: skipped, TotalBlocks: blockCount})
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "xfer", Copied: copied, Skipped: skipped, SrcRead: srcRead, TotalBlocks: blockCount})
|
||||
}
|
||||
|
||||
loopErr := runSourceLoop(sourceLoopParams{
|
||||
File: f, Base: base, Size: size, BlockSize: blockSize, Align: align, Hashes: hashCh, Out: out,
|
||||
File: f, Size: size, BlockSize: blockSize, Align: align, Hashes: hashCh, Out: out,
|
||||
OnRead: func(uint64) { srcRead++; maybeProgress() },
|
||||
OnSkip: func(uint64) { skipped++; maybeProgress() },
|
||||
OnSend: func(uint64, [32]byte) { copied++; maybeProgress() },
|
||||
})
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "xfer", Copied: copied, Skipped: skipped, SrcRead: srcRead, TotalBlocks: blockCount})
|
||||
if readErr := <-readErrCh; readErr != nil {
|
||||
return readErr
|
||||
}
|
||||
|
||||
5
build.sh
5
build.sh
@ -4,10 +4,13 @@ set -e
|
||||
cd "$(dirname "$0")"
|
||||
mkdir -p dist
|
||||
|
||||
build_time=$(date -u +%Y-%m-%dT%H:%M:%SZ)
|
||||
|
||||
build() {
|
||||
os=$1 arch=$2 out=$3
|
||||
echo "==> $os/$arch -> dist/$out"
|
||||
CGO_ENABLED=0 GOOS="$os" GOARCH="$arch" go build -trimpath -ldflags "-s -w" -o "dist/$out" .
|
||||
CGO_ENABLED=0 GOOS="$os" GOARCH="$arch" go build -trimpath \
|
||||
-ldflags "-s -w -X main.buildTime=$build_time" -o "dist/$out" .
|
||||
}
|
||||
|
||||
build linux amd64 clonetool-linux-amd64
|
||||
|
||||
18
ctrlmsg.go
18
ctrlmsg.go
@ -48,13 +48,23 @@ type CtrlMsg struct {
|
||||
Reason string `json:"reason,omitempty"`
|
||||
Message string `json:"message,omitempty"`
|
||||
|
||||
// progress. Scan and transfer now overlap, so a run emits both "scan"
|
||||
// messages (Hashed/TotalBlocks) and transfer messages (Copied/Skipped)
|
||||
// interleaved; the printer keeps the latest of each.
|
||||
Phase string `json:"phase,omitempty"` // "scan" while hashing the destination; empty during transfer
|
||||
// progress. Scan, source read and destination write all overlap and are
|
||||
// tracked independently, so a run emits several kinds of "progress"
|
||||
// message distinguished by Phase, interleaved; the printer keeps the
|
||||
// latest of each:
|
||||
// "scan" - Hashed/TotalBlocks: destination read throughput while it
|
||||
// fingerprints its current content.
|
||||
// "xfer" - Copied/Skipped/SrcRead/TotalBlocks: source read
|
||||
// throughput as it reads and compares each block (Skipped:
|
||||
// matched the destination; Copied: differed and was sent).
|
||||
// "write" - Written: destination write throughput as changed blocks
|
||||
// are actually applied.
|
||||
Phase string `json:"phase,omitempty"`
|
||||
Hashed int64 `json:"hashed,omitempty"`
|
||||
Copied int64 `json:"copied,omitempty"`
|
||||
Skipped int64 `json:"skipped,omitempty"`
|
||||
SrcRead int64 `json:"srcRead,omitempty"`
|
||||
Written int64 `json:"written,omitempty"`
|
||||
TotalBlocks int64 `json:"totalBlocks,omitempty"`
|
||||
BytesCopied int64 `json:"bytesCopied,omitempty"`
|
||||
BytesTotal int64 `json:"bytesTotal,omitempty"`
|
||||
|
||||
29
deploy.go
29
deploy.go
@ -15,26 +15,43 @@ import (
|
||||
// no install, no root.
|
||||
const deployedRemoteBin = ".clonetool/bin/clonetool"
|
||||
|
||||
// resolveRemoteBin returns the path to a runnable clonetool on spec's host.
|
||||
// It checks the configured --remote-bin first, then a copy deployed by an
|
||||
// earlier run, and finally streams this binary over and verifies it runs.
|
||||
// resolveRemoteBin returns the path to a clonetool on spec's host that is
|
||||
// both runnable and the same build as this binary (compared via the
|
||||
// "build=" tag in `version` output, see versionString/remoteBuildTag — an
|
||||
// architecture mismatch alone, expected when cross-compiled, doesn't count
|
||||
// as a mismatch). It checks the configured --remote-bin first, then a copy
|
||||
// deployed by an earlier run, and finally streams this binary over and
|
||||
// verifies it runs. With --deploy=false it can't correct a stale or missing
|
||||
// binary, so it uses --remote-bin anyway if merely out of date (only a
|
||||
// non-runnable one is still an error).
|
||||
func resolveRemoteBin(cfg *SyncConfig, spec Spec, tag string) (string, error) {
|
||||
if _, code, err := runRemote(cfg, spec, cfg.RemoteBin, "version"); err != nil {
|
||||
localTag := remoteBuildTag(versionString())
|
||||
|
||||
if out, code, err := runRemote(cfg, spec, cfg.RemoteBin, "version"); err != nil {
|
||||
return "", fmt.Errorf("%s: cannot reach %s over ssh: %w", tag, spec.Host, err)
|
||||
} else if code == 0 {
|
||||
if remoteBuildTag(out) == localTag {
|
||||
return cfg.RemoteBin, nil
|
||||
}
|
||||
if !cfg.Deploy {
|
||||
fmt.Fprintf(os.Stderr, "%s: %s on %s reports %q, a different build than this one (%s); continuing anyway (--deploy=false)\n",
|
||||
tag, cfg.RemoteBin, spec.Host, firstLine(out), versionString())
|
||||
return cfg.RemoteBin, nil
|
||||
}
|
||||
fmt.Fprintf(os.Stderr, "%s: %s on %s reports %q, a different build than this one (%s); looking for an up-to-date copy ...\n",
|
||||
tag, cfg.RemoteBin, spec.Host, firstLine(out), versionString())
|
||||
}
|
||||
|
||||
if !cfg.Deploy {
|
||||
return "", fmt.Errorf("%s: %q is not runnable on %s and --deploy=false; "+
|
||||
"install clonetool there or pass --remote-bin", tag, cfg.RemoteBin, spec.Host)
|
||||
}
|
||||
|
||||
if _, code, err := runRemote(cfg, spec, deployedRemoteBin, "version"); err == nil && code == 0 {
|
||||
if out, code, err := runRemote(cfg, spec, deployedRemoteBin, "version"); err == nil && code == 0 && remoteBuildTag(out) == localTag {
|
||||
return deployedRemoteBin, nil
|
||||
}
|
||||
|
||||
fmt.Fprintf(os.Stderr, "%s: no runnable clonetool on %s; copying this binary there ...\n", tag, spec.Host)
|
||||
fmt.Fprintf(os.Stderr, "%s: no up-to-date clonetool on %s; copying this binary there ...\n", tag, spec.Host)
|
||||
if err := deployBinary(cfg, spec); err != nil {
|
||||
return "", fmt.Errorf("%s: copy clonetool to %s: %w", tag, spec.Host, err)
|
||||
}
|
||||
|
||||
4
main.go
4
main.go
@ -4,8 +4,6 @@ import (
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"runtime"
|
||||
"strings"
|
||||
)
|
||||
|
||||
func main() {
|
||||
@ -23,7 +21,7 @@ func main() {
|
||||
case "agent":
|
||||
err = cmdAgent(os.Args[2:])
|
||||
case "version", "--version":
|
||||
fmt.Printf("clonetool %s/%s\n", runtime.GOOS, runtime.GOARCH)
|
||||
fmt.Println(versionString())
|
||||
return
|
||||
case "-h", "--help", "help":
|
||||
usage()
|
||||
|
||||
82
manager.go
82
manager.go
@ -234,28 +234,62 @@ func confirmShrink(dstSpec Spec, oldSize, newSize int64) bool {
|
||||
return line == "y" || line == "yes"
|
||||
}
|
||||
|
||||
// progressPrinter renders a single rewriting status line on stderr. Scan
|
||||
// and transfer now run concurrently, so it keeps the latest of each kind of
|
||||
// update (blocks hashed so far, blocks copied/skipped) and shows them on one
|
||||
// line, with a rolling copy rate; the "scan H/M" segment disappears once the
|
||||
// rateTracker smooths a monotonically increasing block counter into a
|
||||
// blocks/sec rate, sampling no more than a few times a second so a burst of
|
||||
// same-tick progress messages doesn't produce a noisy instantaneous rate.
|
||||
type rateTracker struct {
|
||||
lastT time.Time
|
||||
lastVal int64
|
||||
rate float64
|
||||
}
|
||||
|
||||
func (r *rateTracker) update(val int64) {
|
||||
now := time.Now()
|
||||
if r.lastT.IsZero() {
|
||||
r.lastT, r.lastVal = now, val
|
||||
return
|
||||
}
|
||||
dt := now.Sub(r.lastT).Seconds()
|
||||
if dt < 0.4 {
|
||||
return
|
||||
}
|
||||
inst := float64(val-r.lastVal) / dt
|
||||
if r.rate == 0 {
|
||||
r.rate = inst
|
||||
} else {
|
||||
r.rate = 0.6*r.rate + 0.4*inst
|
||||
}
|
||||
r.lastT, r.lastVal = now, val
|
||||
}
|
||||
|
||||
func (r *rateTracker) mib(blockSize int64) float64 {
|
||||
return r.rate * float64(blockSize) / (1024 * 1024)
|
||||
}
|
||||
|
||||
// progressPrinter renders a single rewriting status line on stderr. Scan,
|
||||
// source read and destination write all run concurrently, each tracked by
|
||||
// its own phase-tagged progress message (see CtrlMsg.Phase), so it keeps the
|
||||
// latest of each and shows them on one line with their own rolling
|
||||
// throughput; the "scan H/M" segment and its read rate disappear once the
|
||||
// destination fingerprint is complete.
|
||||
type progressPrinter struct {
|
||||
blockSize int64
|
||||
start time.Time
|
||||
lastT time.Time
|
||||
lastCopied int64
|
||||
rate float64 // copied blocks/sec, smoothed
|
||||
active bool
|
||||
|
||||
hashed int64
|
||||
copied int64
|
||||
skipped int64
|
||||
srcRead int64
|
||||
written int64
|
||||
total int64
|
||||
|
||||
dstReadRate rateTracker // destination scan/hash throughput
|
||||
srcReadRate rateTracker // source read/compare throughput
|
||||
dstWriteRate rateTracker // destination write throughput
|
||||
}
|
||||
|
||||
func newProgressPrinter(blockSize int64) *progressPrinter {
|
||||
now := time.Now()
|
||||
return &progressPrinter{blockSize: blockSize, start: now, lastT: now}
|
||||
return &progressPrinter{blockSize: blockSize}
|
||||
}
|
||||
|
||||
func (p *progressPrinter) print(m CtrlMsg) {
|
||||
@ -263,22 +297,22 @@ func (p *progressPrinter) print(m CtrlMsg) {
|
||||
if m.TotalBlocks > 0 {
|
||||
p.total = m.TotalBlocks
|
||||
}
|
||||
if m.Phase == "scan" {
|
||||
switch m.Phase {
|
||||
case "scan":
|
||||
if m.Hashed > p.hashed {
|
||||
p.hashed = m.Hashed
|
||||
}
|
||||
if p.total > 0 && p.hashed >= p.total {
|
||||
p.dstReadRate.rate = 0 // scan finished; stop showing a frozen rate
|
||||
} else {
|
||||
p.copied, p.skipped = m.Copied, m.Skipped
|
||||
now := time.Now()
|
||||
if dt := now.Sub(p.lastT).Seconds(); dt >= 0.4 {
|
||||
inst := float64(m.Copied-p.lastCopied) / dt
|
||||
if p.rate == 0 {
|
||||
p.rate = inst
|
||||
} else {
|
||||
p.rate = 0.6*p.rate + 0.4*inst
|
||||
}
|
||||
p.lastT, p.lastCopied = now, m.Copied
|
||||
p.dstReadRate.update(p.hashed)
|
||||
}
|
||||
case "xfer":
|
||||
p.copied, p.skipped, p.srcRead = m.Copied, m.Skipped, m.SrcRead
|
||||
p.srcReadRate.update(p.srcRead)
|
||||
case "write":
|
||||
p.written = m.Written
|
||||
p.dstWriteRate.update(p.written)
|
||||
}
|
||||
|
||||
done := p.copied + p.skipped
|
||||
@ -286,15 +320,15 @@ func (p *progressPrinter) print(m CtrlMsg) {
|
||||
if total <= 0 {
|
||||
total = done
|
||||
}
|
||||
mib := p.rate * float64(p.blockSize) / (1024 * 1024)
|
||||
|
||||
scan := ""
|
||||
if p.total > 0 && p.hashed < p.total {
|
||||
scan = fmt.Sprintf(" scan %s/%s", formatCount(p.hashed), formatCount(p.total))
|
||||
}
|
||||
fmt.Fprintf(os.Stderr, "\r syncing %s %s/%s copied=%s skipped=%s%s %6.1f MiB/s ",
|
||||
fmt.Fprintf(os.Stderr, "\r syncing %s %s/%s copied=%s skipped=%s%s rd(src) %5.1f rd(dst) %5.1f wr(dst) %5.1f MiB/s ",
|
||||
progressBar(done, total), formatCount(done), formatCount(total),
|
||||
formatCount(p.copied), formatCount(p.skipped), scan, mib)
|
||||
formatCount(p.copied), formatCount(p.skipped), scan,
|
||||
p.srcReadRate.mib(p.blockSize), p.dstReadRate.mib(p.blockSize), p.dstWriteRate.mib(p.blockSize))
|
||||
}
|
||||
|
||||
// printBytes renders a byte-oriented status line for clone-disk's fs-image
|
||||
|
||||
15
syncside.go
15
syncside.go
@ -139,6 +139,11 @@ type sourceLoopParams struct {
|
||||
Align int64
|
||||
Hashes <-chan blockHash
|
||||
Out *FrameWriter
|
||||
// OnRead is called immediately after a block has been read and hashed
|
||||
// for comparison — whether it turns out to match or differ — so
|
||||
// callers can track true source read throughput independent of the
|
||||
// send/ack timing tracked by OnSkip/OnSend below.
|
||||
OnRead func(index uint64)
|
||||
OnSkip func(index uint64)
|
||||
// OnSend is called for a differing block immediately before its DATA
|
||||
// frame goes on the wire — never after. The push driver relies on this
|
||||
@ -161,6 +166,9 @@ func runSourceLoop(p sourceLoopParams) error {
|
||||
return fmt.Errorf("read %s at block %d: %w", f.Name(), bh.index, err)
|
||||
}
|
||||
hash := sha256.Sum256(buf[:n])
|
||||
if p.OnRead != nil {
|
||||
p.OnRead(bh.index)
|
||||
}
|
||||
if hash == bh.hash {
|
||||
if p.OnSkip != nil {
|
||||
p.OnSkip(bh.index)
|
||||
@ -194,6 +202,10 @@ type destLoopParams struct {
|
||||
// pull mode, where the remote source-stream has no other channel back
|
||||
// to the manager for progress updates). Sink-role callers leave it nil.
|
||||
OnCtrlMsg func(CtrlMsg)
|
||||
// OnWrite is called right after a block is durably written — the true
|
||||
// destination write-throughput measurement point, as opposed to
|
||||
// whatever "copied" count a driver derives from the send/ack protocol.
|
||||
OnWrite func(index uint64)
|
||||
}
|
||||
|
||||
func runDestLoop(p destLoopParams) error {
|
||||
@ -223,6 +235,9 @@ func runDestLoop(p destLoopParams) error {
|
||||
}
|
||||
return fmt.Errorf("write block %d: %w", index, err)
|
||||
}
|
||||
if p.OnWrite != nil {
|
||||
p.OnWrite(index)
|
||||
}
|
||||
if p.AckOut != nil {
|
||||
if err := p.AckOut.WriteFrame(frameAck, encodeIndexFrame(index)); err != nil {
|
||||
return err
|
||||
|
||||
35
version.go
Normal file
35
version.go
Normal file
@ -0,0 +1,35 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"runtime"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// buildTime is stamped at build time via -ldflags "-X main.buildTime=..."
|
||||
// (build.sh does this). A plain "go build" with no ldflags leaves it at
|
||||
// "dev" — still a meaningful, comparable value: two "dev" builds are only
|
||||
// ever the same binary bytes copied around (self-deploy), never two
|
||||
// independently built ones landing on the same value by chance.
|
||||
var buildTime = "dev"
|
||||
|
||||
// versionString is what `clonetool version` prints, and what a remote
|
||||
// host's own `version` output is compared against (see remoteBuildTag) to
|
||||
// tell whether it's running the same build as this binary.
|
||||
func versionString() string {
|
||||
return fmt.Sprintf("clonetool %s/%s build=%s", runtime.GOOS, runtime.GOARCH, buildTime)
|
||||
}
|
||||
|
||||
// remoteBuildTag extracts the "build=..." tag from a clonetool `version`
|
||||
// line, ignoring its GOOS/GOARCH — those are expected to legitimately
|
||||
// differ across a cross-compiled deploy — and any surrounding ssh/shell
|
||||
// noise on other lines. Returns "" if the line isn't a clonetool version
|
||||
// line at all (e.g. "command not found").
|
||||
func remoteBuildTag(out string) string {
|
||||
line := firstLine(out)
|
||||
i := strings.Index(line, "build=")
|
||||
if i < 0 {
|
||||
return ""
|
||||
}
|
||||
return strings.TrimSpace(line[i+len("build="):])
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user