Initial D-InSAR bundle restore tool
This commit is contained in:
Harmon
@@ -0,0 +1,847 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
pairsFileName = "pairs.json"
|
||||
reportFileName = "restore_report.json"
|
||||
logFileName = "restore.log"
|
||||
pairMetaName = ".dinsar_pair.json"
|
||||
tempSuffix = ".tmp"
|
||||
)
|
||||
|
||||
type config struct {
|
||||
inputRoot string
|
||||
outputRoot string
|
||||
skipExisting bool
|
||||
overwrite bool
|
||||
limit int
|
||||
dryRun bool
|
||||
checkOnly bool
|
||||
logWriter io.Writer
|
||||
}
|
||||
|
||||
type pairsDocument struct {
|
||||
Schema string `json:"schema"`
|
||||
ExportedAt string `json:"exported_at"`
|
||||
Pairs []pair `json:"pairs"`
|
||||
}
|
||||
|
||||
type pair struct {
|
||||
PairID string `json:"pair_id"`
|
||||
TaskName string `json:"task_name"`
|
||||
TaskAlias string `json:"task_alias"`
|
||||
MasterData string `json:"master_data"`
|
||||
SlaveData string `json:"slave_data"`
|
||||
MasterOrbit string `json:"master_orbit"`
|
||||
SlaveOrbit string `json:"slave_orbit"`
|
||||
MasterImagingDate string `json:"master_imaging_date"`
|
||||
SlaveImagingDate string `json:"slave_imaging_date"`
|
||||
TimeBaselineDays int `json:"time_baseline_days"`
|
||||
}
|
||||
|
||||
type pairMetadata struct {
|
||||
PairID string `json:"pair_id"`
|
||||
TaskName string `json:"task_name"`
|
||||
TaskAlias string `json:"task_alias"`
|
||||
MasterData string `json:"master_data"`
|
||||
SlaveData string `json:"slave_data"`
|
||||
MasterOrbit string `json:"master_orbit"`
|
||||
SlaveOrbit string `json:"slave_orbit"`
|
||||
MasterImagingDate string `json:"master_imaging_date"`
|
||||
SlaveImagingDate string `json:"slave_imaging_date"`
|
||||
TimeBaselineDays int `json:"time_baseline_days"`
|
||||
RestoredAt string `json:"restored_at"`
|
||||
}
|
||||
|
||||
type restoreReport struct {
|
||||
StartedAt string `json:"started_at"`
|
||||
FinishedAt string `json:"finished_at"`
|
||||
InputRoot string `json:"input_root"`
|
||||
OutputRoot string `json:"output_root"`
|
||||
TotalPairs int `json:"total_pairs"`
|
||||
Restored int `json:"restored"`
|
||||
Skipped int `json:"skipped"`
|
||||
Failed int `json:"failed"`
|
||||
Warnings []string `json:"warnings"`
|
||||
}
|
||||
|
||||
type restoreResult struct {
|
||||
report restoreReport
|
||||
}
|
||||
|
||||
type checkReport struct {
|
||||
StartedAt string `json:"started_at"`
|
||||
FinishedAt string `json:"finished_at"`
|
||||
InputRoot string `json:"input_root"`
|
||||
TotalPairs int `json:"total_pairs"`
|
||||
Checked int `json:"checked"`
|
||||
Valid int `json:"valid"`
|
||||
Failed int `json:"failed"`
|
||||
Warnings []string `json:"warnings"`
|
||||
Errors []string `json:"errors"`
|
||||
}
|
||||
|
||||
type checkResult struct {
|
||||
report checkReport
|
||||
}
|
||||
|
||||
func main() {
|
||||
if len(os.Args) == 1 && runtime.GOOS == "windows" {
|
||||
if err := runGUI(); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "error:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
cfg, err := parseFlags(os.Args[1:])
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "error:", err)
|
||||
os.Exit(2)
|
||||
}
|
||||
|
||||
if cfg.checkOnly {
|
||||
result, err := runCheck(cfg)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "error:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
fmt.Printf("check: total=%d checked=%d valid=%d failed=%d warnings=%d\n",
|
||||
result.report.TotalPairs,
|
||||
result.report.Checked,
|
||||
result.report.Valid,
|
||||
result.report.Failed,
|
||||
len(result.report.Warnings),
|
||||
)
|
||||
if result.report.Failed > 0 {
|
||||
os.Exit(1)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
result, err := run(cfg)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "error:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
fmt.Printf("done: restored=%d skipped=%d failed=%d warnings=%d\n",
|
||||
result.report.Restored,
|
||||
result.report.Skipped,
|
||||
result.report.Failed,
|
||||
len(result.report.Warnings),
|
||||
)
|
||||
}
|
||||
|
||||
func parseFlags(args []string) (config, error) {
|
||||
var cfg config
|
||||
fs := flag.NewFlagSet("dinsar-restore", flag.ContinueOnError)
|
||||
fs.SetOutput(io.Discard)
|
||||
fs.StringVar(&cfg.inputRoot, "input", "", "source bundle root")
|
||||
fs.StringVar(&cfg.outputRoot, "output", "", "restored output root")
|
||||
fs.BoolVar(&cfg.skipExisting, "skip-existing", true, "skip existing completed tasks")
|
||||
fs.BoolVar(&cfg.overwrite, "overwrite", false, "delete and rebuild existing task directories")
|
||||
fs.IntVar(&cfg.limit, "limit", 0, "maximum number of pairs to restore")
|
||||
fs.BoolVar(&cfg.dryRun, "dry-run", false, "print plan without copying files")
|
||||
fs.BoolVar(&cfg.checkOnly, "check-only", false, "validate the source bundle without restoring")
|
||||
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return cfg, err
|
||||
}
|
||||
explicitFlags := map[string]bool{}
|
||||
fs.Visit(func(f *flag.Flag) {
|
||||
explicitFlags[f.Name] = true
|
||||
})
|
||||
if cfg.inputRoot == "" {
|
||||
return cfg, errors.New("missing --input")
|
||||
}
|
||||
if cfg.outputRoot == "" && !cfg.checkOnly {
|
||||
return cfg, errors.New("missing --output")
|
||||
}
|
||||
if cfg.overwrite && explicitFlags["skip-existing"] {
|
||||
return cfg, errors.New("--skip-existing and --overwrite cannot be used together")
|
||||
}
|
||||
if cfg.overwrite {
|
||||
cfg.skipExisting = false
|
||||
}
|
||||
if cfg.limit < 0 {
|
||||
return cfg, errors.New("--limit must be zero or greater")
|
||||
}
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
func run(cfg config) (restoreResult, error) {
|
||||
started := time.Now().UTC()
|
||||
|
||||
inputRoot, err := filepath.Abs(cfg.inputRoot)
|
||||
if err != nil {
|
||||
return restoreResult{}, fmt.Errorf("resolve input root: %w", err)
|
||||
}
|
||||
outputRoot, err := filepath.Abs(cfg.outputRoot)
|
||||
if err != nil {
|
||||
return restoreResult{}, fmt.Errorf("resolve output root: %w", err)
|
||||
}
|
||||
|
||||
report := restoreReport{
|
||||
StartedAt: started.Format(time.RFC3339),
|
||||
InputRoot: inputRoot,
|
||||
OutputRoot: outputRoot,
|
||||
Warnings: []string{},
|
||||
}
|
||||
|
||||
pairsDoc, err := loadAndValidateInput(inputRoot)
|
||||
if err != nil {
|
||||
return restoreResult{}, err
|
||||
}
|
||||
report.TotalPairs = len(pairsDoc.Pairs)
|
||||
|
||||
if !cfg.dryRun {
|
||||
if err := os.MkdirAll(outputRoot, 0755); err != nil {
|
||||
return restoreResult{}, fmt.Errorf("create output root: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
logOutput := defaultLogWriter(cfg.logWriter)
|
||||
logger, logFile, err := newLogger(outputRoot, cfg.dryRun, logOutput)
|
||||
if err != nil {
|
||||
return restoreResult{}, err
|
||||
}
|
||||
if logFile != nil {
|
||||
defer logFile.Close()
|
||||
}
|
||||
|
||||
plannedPairs := pairsDoc.Pairs
|
||||
|
||||
logger.Printf("started input=%s output=%s total_pairs=%d planned_pairs=%d dry_run=%t",
|
||||
inputRoot, outputRoot, report.TotalPairs, len(plannedPairs), cfg.dryRun)
|
||||
|
||||
limitCount := 0
|
||||
for i, p := range plannedPairs {
|
||||
if cfg.limit > 0 && limitCount >= cfg.limit {
|
||||
logger.Printf("limit reached processed=%d limit=%d", limitCount, cfg.limit)
|
||||
break
|
||||
}
|
||||
issues := checkPair(inputRoot, p, false)
|
||||
if len(issues.errors) > 0 {
|
||||
report.Failed++
|
||||
for _, item := range issues.errors {
|
||||
logger.Printf("failed pair_index=%d pair_id=%s error=%s", i, p.PairID, item)
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
warnings, err := restorePair(cfg, inputRoot, outputRoot, p, logger)
|
||||
report.Warnings = append(report.Warnings, warnings...)
|
||||
if err != nil {
|
||||
if errors.Is(err, errSkipped) {
|
||||
report.Skipped++
|
||||
continue
|
||||
}
|
||||
report.Failed++
|
||||
logger.Printf("failed pair_index=%d pair_id=%s error=%v", i, p.PairID, err)
|
||||
continue
|
||||
}
|
||||
limitCount++
|
||||
if !cfg.dryRun {
|
||||
report.Restored++
|
||||
}
|
||||
}
|
||||
|
||||
report.FinishedAt = time.Now().UTC().Format(time.RFC3339)
|
||||
logger.Printf("finished restored=%d skipped=%d failed=%d warnings=%d",
|
||||
report.Restored, report.Skipped, report.Failed, len(report.Warnings))
|
||||
|
||||
if !cfg.dryRun {
|
||||
if err := writeJSON(filepath.Join(outputRoot, reportFileName), report); err != nil {
|
||||
return restoreResult{}, fmt.Errorf("write report: %w", err)
|
||||
}
|
||||
} else {
|
||||
if err := json.NewEncoder(logOutput).Encode(report); err != nil {
|
||||
return restoreResult{}, fmt.Errorf("print dry-run report: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
return restoreResult{report: report}, nil
|
||||
}
|
||||
|
||||
func runCheck(cfg config) (checkResult, error) {
|
||||
started := time.Now().UTC()
|
||||
inputRoot, err := filepath.Abs(cfg.inputRoot)
|
||||
if err != nil {
|
||||
return checkResult{}, fmt.Errorf("resolve input root: %w", err)
|
||||
}
|
||||
|
||||
report := checkReport{
|
||||
StartedAt: started.Format(time.RFC3339),
|
||||
InputRoot: inputRoot,
|
||||
Warnings: []string{},
|
||||
Errors: []string{},
|
||||
}
|
||||
|
||||
logOutput := defaultLogWriter(cfg.logWriter)
|
||||
logger := log.New(logOutput, "", log.LstdFlags)
|
||||
logger.Printf("check started input=%s", inputRoot)
|
||||
|
||||
pairsDoc, err := loadAndValidateInput(inputRoot)
|
||||
if err != nil {
|
||||
report.Failed = 1
|
||||
report.Errors = append(report.Errors, err.Error())
|
||||
report.FinishedAt = time.Now().UTC().Format(time.RFC3339)
|
||||
printCheckReport(logOutput, report)
|
||||
return checkResult{report: report}, nil
|
||||
}
|
||||
report.TotalPairs = len(pairsDoc.Pairs)
|
||||
|
||||
plannedPairs := pairsDoc.Pairs
|
||||
if cfg.limit > 0 && cfg.limit < len(plannedPairs) {
|
||||
plannedPairs = plannedPairs[:cfg.limit]
|
||||
}
|
||||
|
||||
seenTasks := map[string]int{}
|
||||
for i, p := range plannedPairs {
|
||||
report.Checked++
|
||||
issues := checkPair(inputRoot, p, true)
|
||||
taskName, taskNameErr := taskDirectoryName(p)
|
||||
if taskNameErr == nil {
|
||||
if firstIndex, exists := seenTasks[taskName]; exists {
|
||||
issues.errors = append(issues.errors, fmt.Sprintf("duplicate task directory %s also used by pair index %d", taskName, firstIndex))
|
||||
} else {
|
||||
seenTasks[taskName] = i
|
||||
}
|
||||
}
|
||||
|
||||
if len(issues.warnings) > 0 {
|
||||
report.Warnings = append(report.Warnings, issues.warnings...)
|
||||
for _, warning := range issues.warnings {
|
||||
logger.Printf("warning pair_index=%d pair_id=%s %s", i, p.PairID, warning)
|
||||
}
|
||||
}
|
||||
if len(issues.errors) > 0 {
|
||||
report.Failed++
|
||||
for _, item := range issues.errors {
|
||||
message := fmt.Sprintf("pair_index=%d pair_id=%s %s", i, p.PairID, item)
|
||||
report.Errors = append(report.Errors, message)
|
||||
logger.Printf("error %s", message)
|
||||
}
|
||||
continue
|
||||
}
|
||||
report.Valid++
|
||||
logger.Printf("valid pair_index=%d pair_id=%s task=%s", i, p.PairID, taskName)
|
||||
}
|
||||
|
||||
report.FinishedAt = time.Now().UTC().Format(time.RFC3339)
|
||||
logger.Printf("check finished total=%d checked=%d valid=%d failed=%d warnings=%d",
|
||||
report.TotalPairs, report.Checked, report.Valid, report.Failed, len(report.Warnings))
|
||||
printCheckReport(logOutput, report)
|
||||
return checkResult{report: report}, nil
|
||||
}
|
||||
|
||||
type pairCheckIssues struct {
|
||||
warnings []string
|
||||
errors []string
|
||||
}
|
||||
|
||||
func checkPair(inputRoot string, p pair, includeOrbitWarnings bool) pairCheckIssues {
|
||||
var issues pairCheckIssues
|
||||
if _, err := taskDirectoryName(p); err != nil {
|
||||
issues.errors = append(issues.errors, err.Error())
|
||||
}
|
||||
if err := validatePairPaths(inputRoot, p); err != nil {
|
||||
issues.errors = append(issues.errors, err.Error())
|
||||
return issues
|
||||
}
|
||||
|
||||
masterSource, _ := safeJoin(inputRoot, p.MasterData)
|
||||
slaveSource, _ := safeJoin(inputRoot, p.SlaveData)
|
||||
if err := requireDir(masterSource, "master_data"); err != nil {
|
||||
issues.errors = append(issues.errors, err.Error())
|
||||
} else if nonEmpty, err := dirNonEmpty(masterSource); err != nil {
|
||||
issues.errors = append(issues.errors, fmt.Sprintf("check master_data contents: %v", err))
|
||||
} else if !nonEmpty {
|
||||
issues.errors = append(issues.errors, fmt.Sprintf("master_data directory is empty: %s", p.MasterData))
|
||||
}
|
||||
if err := requireDir(slaveSource, "slave_data"); err != nil {
|
||||
issues.errors = append(issues.errors, err.Error())
|
||||
} else if nonEmpty, err := dirNonEmpty(slaveSource); err != nil {
|
||||
issues.errors = append(issues.errors, fmt.Sprintf("check slave_data contents: %v", err))
|
||||
} else if !nonEmpty {
|
||||
issues.errors = append(issues.errors, fmt.Sprintf("slave_data directory is empty: %s", p.SlaveData))
|
||||
}
|
||||
if includeOrbitWarnings {
|
||||
issues.warnings = append(issues.warnings, checkOrbitWarnings(inputRoot, p)...)
|
||||
}
|
||||
return issues
|
||||
}
|
||||
|
||||
func checkOrbitWarnings(inputRoot string, p pair) []string {
|
||||
taskName, err := taskDirectoryName(p)
|
||||
if err != nil {
|
||||
taskName = pairLabel(p)
|
||||
}
|
||||
return orbitWarnings(inputRoot, p, taskName)
|
||||
}
|
||||
|
||||
func printCheckReport(writer io.Writer, report checkReport) {
|
||||
if writer == nil {
|
||||
return
|
||||
}
|
||||
encoder := json.NewEncoder(writer)
|
||||
encoder.SetIndent("", " ")
|
||||
_ = encoder.Encode(report)
|
||||
}
|
||||
|
||||
func loadAndValidateInput(inputRoot string) (pairsDocument, error) {
|
||||
if stat, err := os.Stat(inputRoot); err != nil {
|
||||
return pairsDocument{}, fmt.Errorf("input root is not accessible: %w", err)
|
||||
} else if !stat.IsDir() {
|
||||
return pairsDocument{}, fmt.Errorf("input root is not a directory: %s", inputRoot)
|
||||
}
|
||||
|
||||
dataDir := filepath.Join(inputRoot, "data")
|
||||
if stat, err := os.Stat(dataDir); err != nil {
|
||||
return pairsDocument{}, fmt.Errorf("data directory is not accessible: %w", err)
|
||||
} else if !stat.IsDir() {
|
||||
return pairsDocument{}, fmt.Errorf("data path is not a directory: %s", dataDir)
|
||||
}
|
||||
|
||||
pairsPath := filepath.Join(inputRoot, pairsFileName)
|
||||
file, err := os.Open(pairsPath)
|
||||
if err != nil {
|
||||
return pairsDocument{}, fmt.Errorf("open pairs.json: %w", err)
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
var doc pairsDocument
|
||||
decoder := json.NewDecoder(file)
|
||||
decoder.DisallowUnknownFields()
|
||||
if err := decoder.Decode(&doc); err != nil {
|
||||
return pairsDocument{}, fmt.Errorf("parse pairs.json: %w", err)
|
||||
}
|
||||
if len(doc.Pairs) == 0 {
|
||||
return pairsDocument{}, errors.New("pairs.json contains no pairs")
|
||||
}
|
||||
return doc, nil
|
||||
}
|
||||
|
||||
func validatePairPaths(inputRoot string, p pair) error {
|
||||
if strings.TrimSpace(p.MasterData) == "" {
|
||||
return fmt.Errorf("pair %s missing master_data", pairLabel(p))
|
||||
}
|
||||
if strings.TrimSpace(p.SlaveData) == "" {
|
||||
return fmt.Errorf("pair %s missing slave_data", pairLabel(p))
|
||||
}
|
||||
if _, err := safeJoin(inputRoot, p.MasterData); err != nil {
|
||||
return fmt.Errorf("pair %s invalid master_data: %w", pairLabel(p), err)
|
||||
}
|
||||
if _, err := safeJoin(inputRoot, p.SlaveData); err != nil {
|
||||
return fmt.Errorf("pair %s invalid slave_data: %w", pairLabel(p), err)
|
||||
}
|
||||
if strings.TrimSpace(p.MasterOrbit) != "" {
|
||||
if _, err := safeJoin(inputRoot, p.MasterOrbit); err != nil {
|
||||
return fmt.Errorf("pair %s invalid master_orbit: %w", pairLabel(p), err)
|
||||
}
|
||||
}
|
||||
if strings.TrimSpace(p.SlaveOrbit) != "" {
|
||||
if _, err := safeJoin(inputRoot, p.SlaveOrbit); err != nil {
|
||||
return fmt.Errorf("pair %s invalid slave_orbit: %w", pairLabel(p), err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
var errSkipped = errors.New("skipped")
|
||||
|
||||
func restorePair(cfg config, inputRoot string, outputRoot string, p pair, logger *log.Logger) ([]string, error) {
|
||||
taskDirName, err := taskDirectoryName(p)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
taskDir := filepath.Join(outputRoot, taskDirName)
|
||||
if !isSubpath(outputRoot, taskDir) {
|
||||
return nil, fmt.Errorf("task directory escapes output root: %s", taskDirName)
|
||||
}
|
||||
tempDir := taskDir + tempSuffix
|
||||
|
||||
masterSource, _ := safeJoin(inputRoot, p.MasterData)
|
||||
slaveSource, _ := safeJoin(inputRoot, p.SlaveData)
|
||||
|
||||
if err := requireDir(masterSource, "master_data"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := requireDir(slaveSource, "slave_data"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
existingComplete, err := taskComplete(taskDir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if existingComplete && cfg.skipExisting {
|
||||
logger.Printf("skip existing task=%s", taskDirName)
|
||||
return nil, errSkipped
|
||||
}
|
||||
if cfg.dryRun {
|
||||
logger.Printf("dry-run restore task=%s master=%s slave=%s", taskDirName, p.MasterData, p.SlaveData)
|
||||
return orbitWarnings(inputRoot, p, taskDirName), nil
|
||||
}
|
||||
if pathExists(taskDir) && !cfg.overwrite {
|
||||
return nil, fmt.Errorf("task already exists and is not complete; use --overwrite or remove it: %s", taskDir)
|
||||
}
|
||||
if cfg.overwrite {
|
||||
if err := os.RemoveAll(taskDir); err != nil {
|
||||
return nil, fmt.Errorf("remove existing task %s: %w", taskDir, err)
|
||||
}
|
||||
}
|
||||
if pathExists(tempDir) {
|
||||
if err := os.RemoveAll(tempDir); err != nil {
|
||||
return nil, fmt.Errorf("remove stale temp directory %s: %w", tempDir, err)
|
||||
}
|
||||
}
|
||||
|
||||
logger.Printf("restore task=%s", taskDirName)
|
||||
if err := os.MkdirAll(tempDir, 0755); err != nil {
|
||||
return nil, fmt.Errorf("create temp task directory: %w", err)
|
||||
}
|
||||
|
||||
cleanupTemp := true
|
||||
defer func() {
|
||||
if cleanupTemp {
|
||||
_ = os.RemoveAll(tempDir)
|
||||
}
|
||||
}()
|
||||
|
||||
if err := copyDirContents(masterSource, filepath.Join(tempDir, "master")); err != nil {
|
||||
return nil, fmt.Errorf("copy master data: %w", err)
|
||||
}
|
||||
if err := copyDirContents(slaveSource, filepath.Join(tempDir, "slave")); err != nil {
|
||||
return nil, fmt.Errorf("copy slave data: %w", err)
|
||||
}
|
||||
|
||||
warnings := copyOrbitFiles(inputRoot, tempDir, taskDirName, p, logger)
|
||||
|
||||
meta := pairMetadata{
|
||||
PairID: p.PairID,
|
||||
TaskName: p.TaskName,
|
||||
TaskAlias: p.TaskAlias,
|
||||
MasterData: p.MasterData,
|
||||
SlaveData: p.SlaveData,
|
||||
MasterOrbit: p.MasterOrbit,
|
||||
SlaveOrbit: p.SlaveOrbit,
|
||||
MasterImagingDate: p.MasterImagingDate,
|
||||
SlaveImagingDate: p.SlaveImagingDate,
|
||||
TimeBaselineDays: p.TimeBaselineDays,
|
||||
RestoredAt: time.Now().UTC().Format(time.RFC3339),
|
||||
}
|
||||
if err := writeJSON(filepath.Join(tempDir, pairMetaName), meta); err != nil {
|
||||
return warnings, fmt.Errorf("write pair metadata: %w", err)
|
||||
}
|
||||
if err := validateRestoredTask(tempDir); err != nil {
|
||||
return warnings, err
|
||||
}
|
||||
if err := os.Rename(tempDir, taskDir); err != nil {
|
||||
return warnings, fmt.Errorf("publish restored task: %w", err)
|
||||
}
|
||||
cleanupTemp = false
|
||||
return warnings, nil
|
||||
}
|
||||
|
||||
func taskDirectoryName(p pair) (string, error) {
|
||||
for _, candidate := range []string{p.TaskAlias, p.TaskName, p.PairID} {
|
||||
name := strings.TrimSpace(candidate)
|
||||
if name == "" {
|
||||
continue
|
||||
}
|
||||
if name != filepath.Base(name) || strings.ContainsAny(name, `/\:`) {
|
||||
return "", fmt.Errorf("invalid task directory name: %q", name)
|
||||
}
|
||||
return name, nil
|
||||
}
|
||||
return "", errors.New("pair has no task_alias, task_name, or pair_id")
|
||||
}
|
||||
|
||||
func copyOrbitFiles(inputRoot string, tempDir string, taskDirName string, p pair, logger *log.Logger) []string {
|
||||
var warnings []string
|
||||
for _, entry := range []struct {
|
||||
label string
|
||||
value string
|
||||
}{
|
||||
{label: "master_orbit", value: p.MasterOrbit},
|
||||
{label: "slave_orbit", value: p.SlaveOrbit},
|
||||
} {
|
||||
if strings.TrimSpace(entry.value) == "" {
|
||||
continue
|
||||
}
|
||||
source, err := safeJoin(inputRoot, entry.value)
|
||||
if err != nil {
|
||||
warning := fmt.Sprintf("task %s %s invalid: %v", taskDirName, entry.label, err)
|
||||
warnings = append(warnings, warning)
|
||||
logger.Printf("warning %s", warning)
|
||||
continue
|
||||
}
|
||||
stat, err := os.Stat(source)
|
||||
if err != nil {
|
||||
warning := fmt.Sprintf("task %s %s missing: %s", taskDirName, entry.label, entry.value)
|
||||
warnings = append(warnings, warning)
|
||||
logger.Printf("warning %s", warning)
|
||||
continue
|
||||
}
|
||||
if stat.IsDir() {
|
||||
warning := fmt.Sprintf("task %s %s is a directory, expected file: %s", taskDirName, entry.label, entry.value)
|
||||
warnings = append(warnings, warning)
|
||||
logger.Printf("warning %s", warning)
|
||||
continue
|
||||
}
|
||||
orbitDir := filepath.Join(tempDir, "orbit")
|
||||
if err := os.MkdirAll(orbitDir, 0755); err != nil {
|
||||
warning := fmt.Sprintf("task %s cannot create orbit directory: %v", taskDirName, err)
|
||||
warnings = append(warnings, warning)
|
||||
logger.Printf("warning %s", warning)
|
||||
continue
|
||||
}
|
||||
if err := copyFile(source, filepath.Join(orbitDir, filepath.Base(source)), stat.Mode()); err != nil {
|
||||
warning := fmt.Sprintf("task %s cannot copy %s %s: %v", taskDirName, entry.label, entry.value, err)
|
||||
warnings = append(warnings, warning)
|
||||
logger.Printf("warning %s", warning)
|
||||
continue
|
||||
}
|
||||
}
|
||||
return warnings
|
||||
}
|
||||
|
||||
func orbitWarnings(inputRoot string, p pair, taskDirName string) []string {
|
||||
var warnings []string
|
||||
for _, entry := range []struct {
|
||||
label string
|
||||
value string
|
||||
}{
|
||||
{label: "master_orbit", value: p.MasterOrbit},
|
||||
{label: "slave_orbit", value: p.SlaveOrbit},
|
||||
} {
|
||||
if strings.TrimSpace(entry.value) == "" {
|
||||
continue
|
||||
}
|
||||
source, err := safeJoin(inputRoot, entry.value)
|
||||
if err != nil {
|
||||
warnings = append(warnings, fmt.Sprintf("task %s %s invalid: %v", taskDirName, entry.label, err))
|
||||
continue
|
||||
}
|
||||
if stat, err := os.Stat(source); err != nil {
|
||||
warnings = append(warnings, fmt.Sprintf("task %s %s missing: %s", taskDirName, entry.label, entry.value))
|
||||
} else if stat.IsDir() {
|
||||
warnings = append(warnings, fmt.Sprintf("task %s %s is a directory, expected file: %s", taskDirName, entry.label, entry.value))
|
||||
}
|
||||
}
|
||||
return warnings
|
||||
}
|
||||
|
||||
func copyDirContents(sourceDir string, destDir string) error {
|
||||
sourceEntries, err := os.ReadDir(sourceDir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := os.MkdirAll(destDir, 0755); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, entry := range sourceEntries {
|
||||
sourcePath := filepath.Join(sourceDir, entry.Name())
|
||||
destPath := filepath.Join(destDir, entry.Name())
|
||||
info, err := entry.Info()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
mode := info.Mode()
|
||||
switch {
|
||||
case mode&os.ModeSymlink != 0:
|
||||
return fmt.Errorf("symlinks are not supported: %s", sourcePath)
|
||||
case info.IsDir():
|
||||
if err := copyDirContents(sourcePath, destPath); err != nil {
|
||||
return err
|
||||
}
|
||||
case mode.IsRegular():
|
||||
if err := copyFile(sourcePath, destPath, mode); err != nil {
|
||||
return err
|
||||
}
|
||||
default:
|
||||
return fmt.Errorf("unsupported file type: %s", sourcePath)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func copyFile(sourcePath string, destPath string, mode os.FileMode) error {
|
||||
if err := os.MkdirAll(filepath.Dir(destPath), 0755); err != nil {
|
||||
return err
|
||||
}
|
||||
source, err := os.Open(sourcePath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer source.Close()
|
||||
|
||||
dest, err := os.OpenFile(destPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, mode)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := io.Copy(dest, source); err != nil {
|
||||
_ = dest.Close()
|
||||
return err
|
||||
}
|
||||
return dest.Close()
|
||||
}
|
||||
|
||||
func taskComplete(taskDir string) (bool, error) {
|
||||
masterNonEmpty, err := dirNonEmpty(filepath.Join(taskDir, "master"))
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
slaveNonEmpty, err := dirNonEmpty(filepath.Join(taskDir, "slave"))
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return masterNonEmpty && slaveNonEmpty, nil
|
||||
}
|
||||
|
||||
func validateRestoredTask(taskDir string) error {
|
||||
masterNonEmpty, err := dirNonEmpty(filepath.Join(taskDir, "master"))
|
||||
if err != nil {
|
||||
return fmt.Errorf("validate master directory: %w", err)
|
||||
}
|
||||
if !masterNonEmpty {
|
||||
return errors.New("restored Task/master is empty")
|
||||
}
|
||||
slaveNonEmpty, err := dirNonEmpty(filepath.Join(taskDir, "slave"))
|
||||
if err != nil {
|
||||
return fmt.Errorf("validate slave directory: %w", err)
|
||||
}
|
||||
if !slaveNonEmpty {
|
||||
return errors.New("restored Task/slave is empty")
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(taskDir, pairMetaName)); err != nil {
|
||||
return fmt.Errorf("validate pair metadata: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func dirNonEmpty(path string) (bool, error) {
|
||||
entries, err := os.ReadDir(path)
|
||||
if err != nil {
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return false, nil
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
return len(entries) > 0, nil
|
||||
}
|
||||
|
||||
func requireDir(path string, label string) error {
|
||||
stat, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s is not accessible: %w", label, err)
|
||||
}
|
||||
if !stat.IsDir() {
|
||||
return fmt.Errorf("%s is not a directory: %s", label, path)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func writeJSON(path string, value any) error {
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
|
||||
return err
|
||||
}
|
||||
file, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
encoder := json.NewEncoder(file)
|
||||
encoder.SetIndent("", " ")
|
||||
return encoder.Encode(value)
|
||||
}
|
||||
|
||||
func newLogger(outputRoot string, dryRun bool, output io.Writer) (*log.Logger, *os.File, error) {
|
||||
if output == nil {
|
||||
output = io.Discard
|
||||
}
|
||||
if dryRun {
|
||||
return log.New(output, "", log.LstdFlags), nil, nil
|
||||
}
|
||||
if err := os.MkdirAll(outputRoot, 0755); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
logFile, err := os.OpenFile(filepath.Join(outputRoot, logFileName), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
writer := io.MultiWriter(output, logFile)
|
||||
return log.New(writer, "", log.LstdFlags), logFile, nil
|
||||
}
|
||||
|
||||
func defaultLogWriter(writer io.Writer) io.Writer {
|
||||
if writer != nil {
|
||||
return writer
|
||||
}
|
||||
return os.Stdout
|
||||
}
|
||||
|
||||
func safeJoin(root string, relative string) (string, error) {
|
||||
if filepath.IsAbs(relative) {
|
||||
return "", fmt.Errorf("absolute path is not allowed: %s", relative)
|
||||
}
|
||||
cleanRelative := filepath.Clean(relative)
|
||||
if cleanRelative == "." || strings.HasPrefix(cleanRelative, ".."+string(filepath.Separator)) || cleanRelative == ".." {
|
||||
return "", fmt.Errorf("path escapes root: %s", relative)
|
||||
}
|
||||
joined := filepath.Join(root, cleanRelative)
|
||||
if !isSubpath(root, joined) {
|
||||
return "", fmt.Errorf("path escapes root: %s", relative)
|
||||
}
|
||||
return joined, nil
|
||||
}
|
||||
|
||||
func isSubpath(root string, path string) bool {
|
||||
absRoot, err := filepath.Abs(root)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
absPath, err := filepath.Abs(path)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
rel, err := filepath.Rel(absRoot, absPath)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
return rel == "." || (!strings.HasPrefix(rel, ".."+string(filepath.Separator)) && rel != "..")
|
||||
}
|
||||
|
||||
func pathExists(path string) bool {
|
||||
_, err := os.Stat(path)
|
||||
return err == nil
|
||||
}
|
||||
|
||||
func pairLabel(p pair) string {
|
||||
if strings.TrimSpace(p.PairID) != "" {
|
||||
return p.PairID
|
||||
}
|
||||
if strings.TrimSpace(p.TaskName) != "" {
|
||||
return p.TaskName
|
||||
}
|
||||
if strings.TrimSpace(p.TaskAlias) != "" {
|
||||
return p.TaskAlias
|
||||
}
|
||||
return "<unknown>"
|
||||
}
|
||||
Reference in New Issue
Block a user