package mirror import ( "context" "errors" "fmt" "os" "path/filepath" "syscall" "time" "github.com/grmrgecko/repo-sync/fetch" log "github.com/sirupsen/logrus" ) // lockDestination takes an exclusive lock under the destination directory // so overlapping runs, such as from cron, cannot race on the same tree. // The caller closes the returned file to release the lock. func lockDestination(dest string) (*os.File, error) { if err := os.MkdirAll(dest, 0755); err != nil { return nil, err } f, err := os.OpenFile(filepath.Join(dest, fetch.LockFileName), os.O_CREATE|os.O_RDWR, 0644) if err != nil { return nil, err } if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { f.Close() return nil, fmt.Errorf("destination %s is locked by another repo-sync run", dest) } return f, nil } // Summary reports what a whole run transferred, for callers that act on the // result rather than read the log. A repository that failed part way through // still counts what it transferred before failing, as those transfers changed // the destination all the same. type Summary struct { Repositories int Failed int DryRun bool Fetched int64 FetchedBytes int64 Unchanged int64 Pruned int64 Planned int64 PlannedBytes int64 } // Changed reports whether the run altered the destination. A dry run reports // what it would have done in the same counters, so it never counts as a // change. func (s *Summary) Changed() bool { if s.DryRun { return false } return s.Fetched > 0 || s.Pruned > 0 } // add folds one repository's counters into the run totals. func (s *Summary) add(c *fetch.Counters) { s.Fetched += c.Fetched.Load() s.FetchedBytes += c.FetchedBytes.Load() s.Unchanged += c.Unchanged.Load() s.Pruned += c.Pruned.Load() s.Planned += c.Planned.Load() s.PlannedBytes += c.PlannedBytes.Load() } // Sync synchronizes every configured repository URL into the destination, // continuing after per-repository failures so one bad repository does not // block the rest. The returned summary covers every repository reached, // including on the error paths, so a caller can still tell whether a failed // run changed anything. func Sync(ctx context.Context, opts *Options) (*Summary, error) { fetch.Reload() summary := &Summary{DryRun: opts.DryRun} // Hold the destination for the whole run; the lock file itself stays // behind, as removing it would race a waiting run. lock, err := lockDestination(opts.Destination) if err != nil { return summary, err } defer lock.Close() // Expand discovery URLs into concrete repository URLs. repoURLs := opts.URLs if opts.Discover { repoURLs = nil for _, base := range opts.URLs { found, err := discoverRepos(ctx, base, opts.Type, opts.DiscoverDepth) if err != nil { return summary, fmt.Errorf("discover repositories under %s: %w", base, err) } log.WithFields(log.Fields{"url": base, "repositories": len(found)}).Info("Discovered repositories.") repoURLs = append(repoURLs, found...) } } if len(repoURLs) == 0 { return summary, errors.New("no repositories to synchronize") } // Synchronize each repository in turn. Every repository's counters are // folded into the run totals before the next one resets them, whatever // its outcome. for _, repoURL := range repoURLs { if err := ctx.Err(); err != nil { return summary, err } log.WithField("url", repoURL).Info("Synchronizing repository.") fetch.Stats.Reset() start := time.Now() err := syncOne(ctx, repoURL, opts) summary.Repositories++ summary.add(fetch.Stats) if err != nil { summary.Failed++ if errors.Is(err, context.Canceled) { return summary, err } log.WithError(err).WithField("url", repoURL).Error("Repository synchronization failed.") continue } logSummary(repoURL, opts, time.Since(start)) } if summary.Failed > 0 { return summary, fmt.Errorf("%d of %d repositories failed to synchronize", summary.Failed, len(repoURLs)) } return summary, nil } // logSummary reports one repository's transfer activity after a successful // synchronization. func logSummary(repoURL string, opts *Options, elapsed time.Duration) { fields := log.Fields{ "url": repoURL, "fetched": fetch.Stats.Fetched.Load(), "transferred": fetch.FormatBytes(fetch.Stats.FetchedBytes.Load()), "unchanged": fetch.Stats.Unchanged.Load(), "duration": elapsed.Round(time.Millisecond).String(), } if opts.DryRun { fields["would_fetch"] = fetch.Stats.Planned.Load() fields["would_transfer"] = fetch.FormatBytes(fetch.Stats.PlannedBytes.Load()) if opts.Prune { fields["would_prune"] = fetch.Stats.Pruned.Load() } log.WithFields(fields).Info("Dry run complete.") return } if opts.Prune { fields["pruned"] = fetch.Stats.Pruned.Load() } log.WithFields(fields).Info("Repository synchronized.") } // syncOne resolves the source for a single repository URL and runs the // format-specific synchronization. Destination paths derive from the layout // URL, so a mirror list maps to the first mirror's path rather than the // list's own URL. func syncOne(ctx context.Context, repoURL string, opts *Options) error { src, layoutURL, err := resolveSource(ctx, repoURL, opts.Type) if err != nil { return err } switch opts.Type { case RepoRPM: destDir, err := opts.repoDest(layoutURL) if err != nil { return err } log.WithField("destination", destDir).Debug("Resolved repository destination.") return syncRPM(ctx, src, destDir, opts) case RepoDeb: return syncDeb(ctx, src, layoutURL, opts) case RepoArch: destDir, err := opts.repoDest(layoutURL) if err != nil { return err } log.WithField("destination", destDir).Debug("Resolved repository destination.") return syncArch(ctx, src, layoutURL, destDir, opts) case RepoApk: destDir, err := opts.repoDest(layoutURL) if err != nil { return err } log.WithField("destination", destDir).Debug("Resolved repository destination.") return syncApk(ctx, src, destDir, opts) default: return fmt.Errorf("unsupported repository type %q", opts.Type) } } // SyncInto synchronizes one repository into an explicit destination // directory, bypassing URL-derived destination mapping. It exists for the // mirror server, whose destinations follow request paths rather than // upstream URL paths. For deb the destination is the archive root // directory; for other types it is the repository directory itself. // Concurrent callers interleave into fetch.Stats, so the counters carry no // per-repository meaning on this path and are neither reset nor reported. // A trace enabled here still reports correct totals: it accumulates its own // counters for the duration of the crawl rather than reading the shared // collector. func SyncInto(ctx context.Context, typ RepoType, src *fetch.Source, repoURL, dest string, opts *Options) error { // Copy before the deb path pins destBase so the caller's Options is // never mutated, keeping the struct safe to reuse or share. serverOpts := *opts opts = &serverOpts switch typ { case RepoRPM: return syncRPM(ctx, src, dest, opts) case RepoDeb: opts.destBase = dest return syncDeb(ctx, src, repoURL, opts) case RepoArch: return syncArch(ctx, src, repoURL, dest, opts) case RepoApk: return syncApk(ctx, src, dest, opts) default: return fmt.Errorf("unsupported repository type %q", typ) } }