proyectos/move-data-nas
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Bump version to 1.0.51

Browse Source
This commit is contained in:
Daniel Arroyo
2026-07-19 19:49:00 -04:00
parent 4ccf2fc2d6
commit 6b29a4b419
11 changed files with 454 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/dto.go
+2
View File
@@ -84,6 +84,8 @@ type JobResponse struct {
ErrorCode *string `json:"error_code,omitempty"`
DurationSeconds *int64 `json:"duration_seconds,omitempty"`
LogLineCount *int64 `json:"log_line_count,omitempty"`
TotalSizeBytes *int64 `json:"total_size_bytes,omitempty"`
SentBytes *int64 `json:"sent_bytes,omitempty"`
}
type LogLineResponse struct {
internal/api/handlers_jobs.go
+6
View File
@@ -254,6 +254,12 @@ func jobToResp(j models.Job) JobResponse {
s := j.FinishedAt.Format(time.RFC3339)
resp.FinishedAt = &s
}
if j.TotalSizeBytes > 0 {
resp.TotalSizeBytes = &j.TotalSizeBytes
}
if j.SentBytes > 0 {
resp.SentBytes = &j.SentBytes
}
return resp
}
internal/db/migrations/0006_progress_totals.sql
+4
View File
@@ -0,0 +1,4 @@
-- 0006_progress_totals.sql
ALTER TABLE jobs ADD COLUMN total_size_bytes INTEGER DEFAULT 0;
ALTER TABLE jobs ADD COLUMN sent_bytes INTEGER DEFAULT 0;
internal/models/job.go
+28 -16
View File
@@ -6,16 +6,18 @@ import (
)
type Job struct {
ID int64 `db:"id" json:"id"`
SyncPairID int64 `db:"sync_pair_id" json:"sync_pair_id"`
TriggerType string `db:"trigger_type" json:"trigger_type"`
Status string `db:"status" json:"status"`
StartedAt *time.Time `db:"started_at" json:"started_at"`
FinishedAt *time.Time `db:"finished_at" json:"finished_at"`
LogFile *string `db:"log_file" json:"log_file"`
ErrorMessage *string `db:"error_message" json:"error_message,omitempty"`
ErrorCode *string `db:"error_code" json:"error_code,omitempty"`
CreatedAt time.Time `db:"created_at" json:"created_at"`
ID int64 `db:"id" json:"id"`
SyncPairID int64 `db:"sync_pair_id" json:"sync_pair_id"`
TriggerType string `db:"trigger_type" json:"trigger_type"`
Status string `db:"status" json:"status"`
StartedAt *time.Time `db:"started_at" json:"started_at"`
FinishedAt *time.Time `db:"finished_at" json:"finished_at"`
LogFile *string `db:"log_file" json:"log_file"`
ErrorMessage *string `db:"error_message" json:"error_message,omitempty"`
ErrorCode *string `db:"error_code" json:"error_code,omitempty"`
TotalSizeBytes int64 `db:"total_size_bytes" json:"total_size_bytes,omitempty"`
SentBytes int64 `db:"sent_bytes" json:"sent_bytes,omitempty"`
CreatedAt time.Time `db:"created_at" json:"created_at"`
}
type JobRepository struct {
@@ -43,9 +45,10 @@ func (r *JobRepository) GetByID(id int64) (*Job, error) {
var logFile, errMsg, errCode sql.NullString
err := r.db.QueryRow(`
SELECT id, sync_pair_id, trigger_type, status, started_at, finished_at,
log_file, error_message, error_code, created_at FROM jobs WHERE id = ?`, id).Scan(
log_file, error_message, error_code, total_size_bytes, sent_bytes, created_at
FROM jobs WHERE id = ?`, id).Scan(
&j.ID, &j.SyncPairID, &j.TriggerType, &j.Status, &started, &finished,
&logFile, &errMsg, &errCode, &j.CreatedAt)
&logFile, &errMsg, &errCode, &j.TotalSizeBytes, &j.SentBytes, &j.CreatedAt)
if err != nil {
return nil, err
}
@@ -70,7 +73,7 @@ func (r *JobRepository) GetByID(id int64) (*Job, error) {
func (r *JobRepository) GetAll(limit, offset int) ([]Job, error) {
rows, err := r.db.Query(`
SELECT id, sync_pair_id, trigger_type, status, started_at, finished_at,
log_file, error_message, error_code, created_at
log_file, error_message, error_code, total_size_bytes, sent_bytes, created_at
FROM jobs ORDER BY created_at DESC LIMIT ? OFFSET ?`,
limit, offset)
if err != nil {
@@ -84,7 +87,8 @@ func (r *JobRepository) GetAll(limit, offset int) ([]Job, error) {
var started, finished sql.NullTime
var logFile, errMsg, errCode sql.NullString
if err := rows.Scan(&j.ID, &j.SyncPairID, &j.TriggerType, &j.Status,
&started, &finished, &logFile, &errMsg, &errCode, &j.CreatedAt); err != nil {
&started, &finished, &logFile, &errMsg, &errCode,
&j.TotalSizeBytes, &j.SentBytes, &j.CreatedAt); err != nil {
return nil, err
}
if started.Valid {
@@ -144,11 +148,11 @@ func (r *JobRepository) GetRunningBySyncPair(syncPairID int64) (*Job, error) {
var logFile, errMsg, errCode sql.NullString
err := r.db.QueryRow(`
SELECT id, sync_pair_id, trigger_type, status, started_at, finished_at,
log_file, error_message, error_code, created_at FROM jobs
log_file, error_message, error_code, total_size_bytes, sent_bytes, created_at FROM jobs
WHERE sync_pair_id = ? AND status IN ('queued','waking_up','running')
ORDER BY created_at DESC LIMIT 1`, syncPairID).Scan(
&j.ID, &j.SyncPairID, &j.TriggerType, &j.Status, &started,
&j.FinishedAt, &logFile, &errMsg, &errCode, &j.CreatedAt)
&j.FinishedAt, &logFile, &errMsg, &errCode, &j.TotalSizeBytes, &j.SentBytes, &j.CreatedAt)
if err != nil {
return nil, err
}
@@ -180,3 +184,11 @@ func (r *JobRepository) DeleteFinishedBefore(before time.Time) (int64, error) {
}
return res.RowsAffected()
}
func (r *JobRepository) SetTotals(id int64, totalSize, sentBytes int64) error {
_, err := r.db.Exec(
"UPDATE jobs SET total_size_bytes = ?, sent_bytes = ? WHERE id = ?",
totalSize, sentBytes, id,
)
return err
}
internal/models/job_log.go
+15 -2
View File
@@ -85,6 +85,19 @@ func (r *JobLogRepository) DeleteBefore(before time.Time) (int64, error) {
return res.RowsAffected()
}
func (r *JobLogRepository) TruncateKeepingHeaderTail(jobID int64, head, tail int) error {
_, err := r.db.Exec(`
DELETE FROM job_logs
WHERE job_id = ?
AND id NOT IN (
SELECT id FROM job_logs WHERE job_id = ? ORDER BY id ASC LIMIT ?
UNION ALL
SELECT id FROM job_logs WHERE job_id = ? ORDER BY id DESC LIMIT ?
)`,
jobID, jobID, head, jobID, tail)
return err
}
type JobWithStats struct {
Job
DurationSeconds *int64 `db:"duration_seconds" json:"duration_seconds"`
@@ -125,7 +138,7 @@ func (r *JobLogRepository) GetAllFiltered(limit, offset int, syncPairID *int64,
SELECT
j.id, j.sync_pair_id, j.trigger_type, j.status,
j.started_at, j.finished_at, j.log_file,
j.error_message, j.error_code, j.created_at,
j.error_message, j.error_code, j.total_size_bytes, j.sent_bytes, j.created_at,
CASE WHEN j.finished_at IS NOT NULL AND j.started_at IS NOT NULL
THEN (j.finished_at - j.started_at) ELSE NULL END as duration_seconds,
(SELECT COUNT(*) FROM job_logs WHERE job_id = j.id) as log_line_count
@@ -148,7 +161,7 @@ func (r *JobLogRepository) GetAllFiltered(limit, offset int, syncPairID *int64,
var durationSeconds sql.NullInt64
if err := rows.Scan(&j.ID, &j.SyncPairID, &j.TriggerType, &j.Status,
&started, &finished, &logFile,
&errMsg, &errCode, &j.CreatedAt,
&errMsg, &errCode, &j.TotalSizeBytes, &j.SentBytes, &j.CreatedAt,
&durationSeconds, &j.LogLineCount); err != nil {
return nil, 0, err
}
internal/syncengine/engine.go
+47 -7
View File
@@ -28,13 +28,16 @@ type Engine struct {
}
type Event struct {
Type string
JobID int64
MachineID int64
Key string
Value string
Line string
Stream string
Type string
JobID int64
MachineID int64
Key string
Value string
Line string
Stream string
Progress *ProgressFields
TotalBytes int64
SentBytes int64
}
func New(database interface{ SQLDB() *sql.DB }, cfg *config.Config) *Engine {
@@ -243,7 +246,17 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error {
}
}
var lastFileName string
onLine := func(stream, line string) {
if stream == "stdout" && isProgressOnlyLine(line) {
if p := parseProgressFields(line); p != nil {
e.emit(Event{Type: "progress", JobID: jobID, Line: line, Stream: stream, Progress: p, Value: lastFileName})
}
return
}
if stream == "stdout" && isFileNameLine(line) {
lastFileName = line
}
f, _ := os.OpenFile(logPath, os.O_APPEND|os.O_WRONLY, 0644)
if f != nil {
fmt.Fprintln(f, line)
@@ -286,6 +299,11 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error {
}
flush()
stats := result.Stats
if stats != nil && stats.TotalSize > 0 {
e.emit(Event{Type: "progress_total", JobID: jobID, TotalBytes: stats.TotalSize, SentBytes: stats.SentBytes})
}
if err != nil {
if jobCtx.Err() != nil {
code := "cancelled_shutdown"
@@ -297,6 +315,7 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error {
e.setJobError(jobID, code, msg)
e.setJobStatus(jobID, "cancelled")
e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "cancelled", Line: msg})
e.persistAndClose(jobID)
return jobCtx.Err()
}
e.setJobStatus(jobID, "failed")
@@ -315,11 +334,15 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error {
e.setJobStatus(jobID, "failed")
e.setJobError(jobID, errCode, errMsg)
e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "failed", Line: errMsg})
e.persistAndClose(jobID)
return fmt.Errorf("rsync exited with code %d: %s", result.ExitCode, result.Stderr)
}
e.setJobStatus(jobID, "success")
e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "success"})
if stats != nil {
e.persistJobTotals(jobID, stats)
}
slog.Info("job completed", "job_id", jobID, "pair", pair.Name)
e.persistAndClose(jobID)
return nil
@@ -327,6 +350,23 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error {
func (e *Engine) persistAndClose(jobID int64) {
e.eventBus.CloseJobChannels(jobID)
logRepo := models.NewJobLogRepository(e.db)
count, err := logRepo.CountByJobID(jobID)
if err == nil && count > 2000 {
if truncateErr := logRepo.TruncateKeepingHeaderTail(jobID, 50, 100); truncateErr != nil {
slog.Warn("failed to truncate job logs", "job_id", jobID, "error", truncateErr)
}
}
}
func (e *Engine) persistJobTotals(jobID int64, stats *RsyncStats) {
if stats == nil {
return
}
jobRepo := models.NewJobRepository(e.db)
if err := jobRepo.SetTotals(jobID, stats.TotalSize, stats.SentBytes); err != nil {
slog.Warn("failed to persist job totals", "job_id", jobID, "error", err)
}
}
func (e *Engine) resolveSSHKey(machine *models.Machine) (string, error) {
internal/syncengine/parser.go
+95
View File
@@ -24,6 +24,15 @@ type ProgressLine struct {
XferedBytes int64
}
type ProgressFields struct {
FileBytes int64
Pct int
SpeedBps int64
EtaSeconds int
XfrDone int
XfrTotal int
}
var (
progressRegex = regexp.MustCompile(`\s*([\d,]+)\s+([\d,]+)\s+([\d%]+)\s*`)
sentRegex = regexp.MustCompile(`sent\s+([\d,]+)\s+bytes`)
@@ -32,6 +41,24 @@ var (
filesRegex = regexp.MustCompile(`Number of files: ([\d,]+)`)
)
var perFileProgressRegex = regexp.MustCompile(
`^\s*(\d{1,3}(?:,\d{3})+)\s+(\d+)%\s+(\d+\.\d+)([kMG])B/s\s+(\d+:\d{2}:\d{2})(.*)`,
)
var xfrRegex = regexp.MustCompile(`xfr#(\d+).*to-chk=(\d+)/(\d+)`)
func parseXfrSuffix(suffix string) (done, total int) {
m := xfrRegex.FindStringSubmatch(suffix)
if m == nil {
return 0, 0
}
done, _ = strconv.Atoi(m[1])
t, _ := strconv.Atoi(m[2])
_ = t
total, _ = strconv.Atoi(m[3])
return done, total
}
func ParseProgressLine(line string) *ProgressLine {
if strings.Contains(line, "files to consider") || strings.Contains(line, "files...") {
return &ProgressLine{Phase: "scanning"}
@@ -74,3 +101,71 @@ func ParseFinalStats(output string) *RsyncStats {
}
return stats
}
func isProgressOnlyLine(line string) bool {
return perFileProgressRegex.MatchString(line)
}
func parseProgressFields(line string) *ProgressFields {
m := perFileProgressRegex.FindStringSubmatch(line)
if m == nil {
return nil
}
bytes, _ := strconv.ParseInt(strings.ReplaceAll(m[1], ",", ""), 10, 64)
pct, _ := strconv.Atoi(m[2])
speed, _ := strconv.ParseFloat(m[3], 64)
unit := m[4]
eta := m[5]
suffix := m[6]
speedBps := int64(speed * 1e6)
switch unit {
case "k", "K":
speedBps = int64(speed * 1e3)
case "m", "M":
speedBps = int64(speed * 1e6)
case "g", "G":
speedBps = int64(speed * 1e9)
}
etaSecs := 0
parts := strings.Split(eta, ":")
if len(parts) == 3 {
h, _ := strconv.Atoi(parts[0])
m, _ := strconv.Atoi(parts[1])
s, _ := strconv.Atoi(parts[2])
etaSecs = h*3600 + m*60 + s
}
pf := &ProgressFields{
FileBytes: bytes,
Pct: pct,
SpeedBps: speedBps,
EtaSeconds: etaSecs,
}
if suffix != "" {
done, total := parseXfrSuffix(suffix)
pf.XfrDone = done
pf.XfrTotal = total
}
return pf
}
func isFileNameLine(line string) bool {
if line == "" || strings.TrimSpace(line) == "" {
return false
}
if strings.Contains(line, "sending incremental file list") ||
strings.Contains(line, "building file list") ||
strings.Contains(line, "cannot open") ||
strings.Contains(line, "skipping non-regular") ||
strings.HasPrefix(line, "sent ") ||
strings.HasPrefix(line, "total ") ||
strings.HasPrefix(line, "Number of files:") ||
strings.Contains(line, "bytes received") {
return false
}
return !isProgressOnlyLine(line)
}
internal/syncengine/parser_test.go
+124
View File
@@ -0,0 +1,124 @@
package syncengine
import "testing"
func TestIsProgressOnlyLine(t *testing.T) {
cases := []struct {
name string
line string
expect bool
}{
{"per-file progress 0%", " 32,768 0% 0.00kB/s 0:00:00", true},
{"per-file progress 7%", " 2,260,893,696 7% 51.14MB/s 0:08:45", true},
{"per-file progress with xfr suffix", " 67,141,632 0% 32.02MB/s 0:15:06 (xfr#1, to-chk=4/10)", true},
{"filename line", "Dragon Ball Sleeping Princess in Devil's Castle (1987)/", false},
{"sending incremental file list header", "sending incremental file list", false},
{"sent bytes stats", "sent 123,456 bytes received 789 bytes 12.34kB/s", false},
{"total size stats", "total size is 999,999,999 speedup is 1.23", false},
{"Number of files stats", "Number of files: 10", false},
{"building file list", "building file list ...", false},
{"empty line", "", false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := isProgressOnlyLine(tc.line)
if got != tc.expect {
t.Errorf("isProgressOnlyLine(%q) = %v, want %v", tc.line, got, tc.expect)
}
})
}
}
func TestParseProgressFields(t *testing.T) {
cases := []struct {
name string
line string
wantPct int
wantSpeedBps int64
wantEtaSeconds int
wantXfrDone int
wantXfrTotal int
}{
{
name: "progress 0% with kB/s",
line: " 32,768 0% 0.00kB/s 0:00:00",
wantPct: 0,
wantSpeedBps: 0,
wantEtaSeconds: 0,
},
{
name: "progress 7% with MB/s",
line: " 2,260,893,696 7% 51.14MB/s 0:08:45",
wantPct: 7,
wantSpeedBps: 51_140_000,
wantEtaSeconds: 8*60 + 45,
},
{
name: "progress with xfr suffix",
line: " 67,141,632 0% 32.02MB/s 0:15:06 (xfr#1, to-chk=4/10)",
wantPct: 0,
wantSpeedBps: 32_020_000,
wantEtaSeconds: 15*60 + 6,
wantXfrDone: 1,
wantXfrTotal: 10,
},
{
name: "progress with GB/s",
line: " 1,234,567,890 50% 1.23GB/s 0:01:30",
wantPct: 50,
wantSpeedBps: 1_230_000_000,
wantEtaSeconds: 1*60 + 30,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
p := parseProgressFields(tc.line)
if p == nil {
t.Fatalf("parseProgressFields(%q) returned nil, want non-nil", tc.line)
}
if p.Pct != tc.wantPct {
t.Errorf("pct = %d, want %d", p.Pct, tc.wantPct)
}
if p.SpeedBps != tc.wantSpeedBps {
t.Errorf("speedBps = %d, want %d", p.SpeedBps, tc.wantSpeedBps)
}
if p.EtaSeconds != tc.wantEtaSeconds {
t.Errorf("etaSeconds = %d, want %d", p.EtaSeconds, tc.wantEtaSeconds)
}
if tc.wantXfrTotal > 0 && p.XfrDone != tc.wantXfrDone {
t.Errorf("xfrDone = %d, want %d", p.XfrDone, tc.wantXfrDone)
}
if tc.wantXfrTotal > 0 && p.XfrTotal != tc.wantXfrTotal {
t.Errorf("xfrTotal = %d, want %d", p.XfrTotal, tc.wantXfrTotal)
}
})
}
}
func TestIsFileNameLine(t *testing.T) {
cases := []struct {
name string
line string
expect bool
}{
{"directory path", "Dragon Ball Sleeping Princess in Devil's Castle (1987)/", true},
{"file path", "Dragon Ball Sleeping Princess in Devil's Castle (1987)/Dragon Ball...WEBDL-2160p.mkv", true},
{"sending incremental file list header", "sending incremental file list", false},
{"sent stats", "sent 12,345 bytes received 1,234 bytes", false},
{"total size stats", "total size is 999,999,999", false},
{"Number of files", "Number of files: 10", false},
{"progress line", " 2,260,893,696 7% 51.14MB/s 0:08:45", false},
{"empty", "", false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := isFileNameLine(tc.line)
if got != tc.expect {
t.Errorf("isFileNameLine(%q) = %v, want %v", tc.line, got, tc.expect)
}
})
}
}