package mirror import ( "context" "fmt" "os" "path/filepath" "sort" "strings" "time" cfg "github.com/grmrgecko/repo-sync/config" "github.com/grmrgecko/repo-sync/fetch" log "github.com/sirupsen/logrus" ) // TraceDir is the directory, relative to a repository, that trace files are // published in. Debian archives established the location and downstream // mirror checkers look for it there. const TraceDir = "project/trace" // TraceInfo describes the mirror a trace file is written for. It carries // the same fields as the configuration's trace section, kept as a separate // type so callers populate it from wherever they hold those details rather // than this package taking a configuration struct as part of its API. type TraceInfo struct { Host string Maintainer string Sponsor string Country string Location string Throughput string } // traceRun records when a repository's synchronization began, and the // transfer it accounted for, so the trace can report both. A nil run writes // nothing, which is how tracing stays off without a condition at every call // site. type traceRun struct { started time.Time stats *fetch.Counters } // newTrace starts timing a repository sync when tracing is enabled. func newTrace(opts *Options) *traceRun { if !opts.Trace { return nil } return &traceRun{started: time.Now(), stats: &fetch.Counters{}} } // track installs the run's own transfer counters on ctx so the trace // reports what this synchronization moved rather than what the shared // collector holds. Without it the mirror server's concurrent crawls would // report each other's bytes. func (t *traceRun) track(ctx context.Context) context.Context { if t == nil { return ctx } return fetch.WithStats(ctx, t.stats) } // publish brings a repository's trace directory up to date: upstream's own // traces are mirrored first, then this run's trace is written, so a name // collision resolves in favour of the file describing this mirror. func (t *traceRun) publish(ctx context.Context, root string, src *fetch.Source, keep *fetch.KeepSet, opts *Options) { if t == nil { return } t.mirrorUpstream(ctx, root, src, keep, opts) t.write(root, src, keep, opts) } // mirrorUpstream copies the traces upstream publishes into this mirror's // trace directory, so a reader can follow the chain back through every // mirror to the archive the content came from. Upstream may publish no // traces at all or may not serve directory indexes, and neither is a // failure of the synchronization, so problems here are logged and the run // continues. func (t *traceRun) mirrorUpstream(ctx context.Context, root string, src *fetch.Source, keep *fetch.KeepSet, opts *Options) { base := src.Active() if base == "" { return } dirURL := base + "/" + TraceDir + "/" listing, err := listDir(ctx, dirURL) if err != nil { log.WithError(err).WithField("url", dirURL).Debug("Upstream serves no trace directory.") return } // Skip the name this mirror traces under; write publishes that file // itself, and upstream's copy of it describes a different mirror. own, err := traceName(opts.TraceInfo.Host) if err != nil { own = "" } names := make([]string, 0, len(listing.files)) for name := range listing.files { if name == own || name == "." || name == ".." { continue } names = append(names, name) } sort.Strings(names) dir := filepath.Join(root, filepath.FromSlash(TraceDir)) var jobs []fetch.Job for _, name := range names { dst, err := fetch.LocalJoin(dir, name) if err != nil { log.WithError(err).WithField("name", name).Debug("Skipping upstream trace with an unusable name.") continue } // Traces carry no checksums, so they revalidate against their // modification time rather than being re-fetched every run, and a // trace withdrawn between the listing and the fetch is not an error. jobs = append(jobs, fetch.Job{ReqPath: TraceDir + "/" + name, Dst: dst, Optional: true}) } if len(jobs) == 0 { return } if opts.DryRun { fetch.PlanJobs(jobs, opts.Verify, keep) return } if _, err := fetch.Many(ctx, src, jobs, opts.Workers, opts.Verify, keep, nil); err != nil { log.WithError(err).WithField("url", dirURL).Warn("Failed to mirror upstream trace files.") return } log.WithFields(log.Fields{"url": dirURL, "traces": len(jobs)}).Debug("Mirrored upstream trace files.") } // traceName returns the file name a mirror is traced under. The value // becomes a file name, so it is reduced to a single path element and // rejected if nothing usable is left. func traceName(host string) (string, error) { name := filepath.Base(strings.TrimSpace(strings.ReplaceAll(host, "\\", "/"))) if name == "." || name == ".." || name == "/" || name == "" { return "", fmt.Errorf("invalid trace host %q", host) } return name, nil } // write publishes the trace for a finished repository synchronization into // root, and records it in keep so pruning does not treat the file as stale // and remove it. On a dry run the file is left untouched but still kept, // so the run does not report the existing trace as about to be pruned. // // A trace describes the mirror rather than the packages, so a failure to // write one is logged and does not fail the synchronization that produced // it. func (t *traceRun) write(root string, src *fetch.Source, keep *fetch.KeepSet, opts *Options) { if t == nil { return } name, err := traceName(opts.TraceInfo.Host) if err != nil { log.WithError(err).Warn("Skipping trace file.") return } dir := filepath.Join(root, filepath.FromSlash(TraceDir)) file := filepath.Join(dir, name) // Keep the trace regardless of whether it is written now, so a dry run // does not report an existing trace as stale. keep.Add(file) if opts.DryRun { return } if err := os.MkdirAll(dir, 0755); err != nil { log.WithError(err).WithField("path", dir).Warn("Failed to create trace directory.") return } content := t.render(src, opts) // Write through a temporary file so a reader never sees a partial // trace, and so an interrupted write cannot truncate the previous one. tmp, err := os.CreateTemp(dir, ".trace-*") if err != nil { log.WithError(err).WithField("path", dir).Warn("Failed to create trace file.") return } tmpName := tmp.Name() if _, err := tmp.WriteString(content); err != nil { tmp.Close() os.Remove(tmpName) log.WithError(err).WithField("path", tmpName).Warn("Failed to write trace file.") return } if err := tmp.Close(); err != nil { os.Remove(tmpName) log.WithError(err).WithField("path", tmpName).Warn("Failed to write trace file.") return } if err := os.Chmod(tmpName, 0644); err != nil { log.WithError(err).WithField("path", tmpName).Debug("Failed to set trace file mode.") } if err := os.Rename(tmpName, file); err != nil { os.Remove(tmpName) log.WithError(err).WithField("path", file).Warn("Failed to publish trace file.") return } log.WithField("path", file).Debug("Wrote trace file.") } // render builds the trace file's contents. The first line is the plain // completion date, as the convention expects, followed by one field per // line. Fields with no configured value are omitted rather than written // empty. func (t *traceRun) render(src *fetch.Source, opts *Options) string { done := time.Now().UTC() var b strings.Builder // The opening line matches date(1)'s default UTC output, which is what // the convention's readers expect; _2 space-pads the day as it does. fmt.Fprintf(&b, "%s\n", done.Format("Mon Jan _2 15:04:05 UTC 2006")) fmt.Fprintf(&b, "Date: %s\n", done.Format(time.RFC1123Z)) fmt.Fprintf(&b, "Date-Started: %s\n", t.started.UTC().Format(time.RFC1123Z)) fmt.Fprintf(&b, "Creator: %s %s\n", cfg.Name, cfg.Version) fmt.Fprintf(&b, "Running on host: %s\n", opts.TraceInfo.Host) for _, f := range []struct{ key, value string }{ {"Maintainer", opts.TraceInfo.Maintainer}, {"Sponsor", opts.TraceInfo.Sponsor}, {"Country", opts.TraceInfo.Country}, {"Location", opts.TraceInfo.Location}, {"Throughput", opts.TraceInfo.Throughput}, } { if f.value != "" { fmt.Fprintf(&b, "%s: %s\n", f.key, f.value) } } fmt.Fprintf(&b, "Repository type: %s\n", opts.Type) if arch := traceArchitectures(opts); arch != "" { fmt.Fprintf(&b, "Architectures: %s\n", arch) } // Report the mirror actually fetched from. For a mirrorlist or metalink // that is the mirror that answered, which is the useful fact and is not // derivable from the configured URL. if src != nil { if active := src.Active(); active != "" { fmt.Fprintf(&b, "Upstream-mirror: %s\n", active) } } elapsed := done.Sub(t.started.UTC()) if elapsed < 0 { elapsed = 0 } seconds := int64(elapsed.Round(time.Second) / time.Second) // The run's own counters, not the shared collector, so a trace written // by one of the server's concurrent crawls reports that crawl alone. bytes := t.stats.FetchedBytes.Load() fmt.Fprintf(&b, "Total bytes received: %d\n", bytes) fmt.Fprintf(&b, "Total time spent syncing: %d\n", seconds) if seconds > 0 { fmt.Fprintf(&b, "Average rate: %d B/s\n", bytes/seconds) } return b.String() } // traceArchitectures reports the architectures a run was restricted to, // which only apt repositories can express. An unrestricted run mirrors // whatever upstream publishes, so no claim is made. func traceArchitectures(opts *Options) string { if opts.Type != RepoDeb || len(opts.Architectures) == 0 { return "" } arch := append([]string(nil), opts.Architectures...) sort.Strings(arch) return strings.Join(arch, " ") }