// seedreplica feeds missing archive revisions to a replica without overwhelming its pull queue.
package main
import (
"bufio"
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"math/rand"
"os"
"os/exec"
"path/filepath"
"regexp"
"runtime"
"sort"
"strconv"
"strings"
"sync"
"text/template"
"time"
"github.com/rcowham/kingpin"
"github.com/sirupsen/logrus"
)
const description = "Seed a replica by verifying missing archive revisions in batches while throttling its pull queue."
const VersionChange = "$Change: 34019 $"
var GoVersion = runtime.Version()
var Version = "dev"
var versionInfoTmpl = `
{{.program}}, version {{.version}} (change: {{.change}})
go version: {{.goVersion}}
`
func VersionPrint(program, version string) string {
change := strings.TrimPrefix(VersionChange, "$Change: ")
change = strings.TrimSuffix(change, " $")
values := map[string]string{
"program": program,
"version": version,
"change": change,
"goVersion": GoVersion,
}
template := template.Must(template.New("version").Parse(versionInfoTmpl))
var output bytes.Buffer
if err := template.ExecuteTemplate(&output, "version", values); err != nil {
panic(err)
}
return strings.TrimSpace(output.String())
}
type config struct {
input string
p4bin string
p4port string
p4user string
batchSize int
maxQueue int
pollInterval time.Duration
maxBatches int
checkpoint string
resume bool
follow bool
dryRun bool
discoverInput string
discoverOnly bool
discoverers int
replaceInput bool
}
type checkpoint struct {
Input string `json:"input"`
Offset int64 `json:"offset"`
LinesRead int64 `json:"lines_read"`
Processed int64 `json:"processed"`
Batches int `json:"batches"`
UpdatedAt time.Time `json:"updated_at"`
CompletedAt time.Time `json:"completed_at,omitempty"`
}
type queueStatus struct {
ActiveTransfers int
TotalTransfers int
ActiveBytes string
TotalBytes string
}
type p4Client struct {
bin string
port string
user string
logger *logrus.Logger
tracking *trackCollector
}
type depot struct {
name string
typeName string
streamDepth int
}
var (
pullQueuePattern = regexp.MustCompile(`(?m)^File transfers:\s*(\d+)\s+active/(\d+)\s+total,\s*bytes:\s*(\S+)\s+active/(\S+)\s+total\.`)
missingRevisionPattern = regexp.MustCompile(`(?i)(//[^#\s]+#\d+)\b.*\b(?:BAD|MISSING)\b`)
sizesFileCountPattern = regexp.MustCompile(`(?i)(\d[\d,]*)\s+files?\b`)
trackLapsePattern = regexp.MustCompile(`^--- lapse ([0-9.]+s)$`)
trackUsagePattern = regexp.MustCompile(`^--- usage ([0-9.]+)\+([0-9.]+)us(?:\s|$)`)
trackTablePattern = regexp.MustCompile(`^--- (db\.\S+)$`)
trackLockPattern = regexp.MustCompile(`^---\s+total lock wait\+held read/write ([0-9.]+)ms\+([0-9.]+)ms/([0-9.]+)ms\+([0-9.]+)ms$`)
trackLibraryPattern = regexp.MustCompile(`^--- lbr (.+)$`)
trackLibraryUsePattern = regexp.MustCompile(`^---\s+opens\+closes\+checkins\+exists (\d+)\+(\d+)\+(\d+)\+(\d+)$`)
)
type tableTrack struct {
ReadWait time.Duration
ReadHeld time.Duration
WriteWait time.Duration
WriteHeld time.Duration
}
type libraryTrack struct {
Opens int64
Closes int64
Checkins int64
Exists int64
}
type trackOutput struct {
Lapse time.Duration
UserCPU time.Duration
SystemCPU time.Duration
Tables map[string]tableTrack
Libraries map[string]libraryTrack
tracked bool
}
type trackCollector struct {
mu sync.Mutex
commands map[string]trackOutput
}
func parseTrackDuration(value string, unit string) time.Duration {
duration, err := time.ParseDuration(value + unit)
if err != nil {
return 0
}
return duration
}
func parseTrackOutput(output string) trackOutput {
track := trackOutput{Tables: make(map[string]tableTrack), Libraries: make(map[string]libraryTrack)}
var table string
var library string
for _, line := range strings.Split(output, "\n") {
line = strings.TrimSpace(line)
if matches := trackLapsePattern.FindStringSubmatch(line); matches != nil {
track.Lapse = parseTrackDuration(matches[1], "")
track.tracked = true
continue
}
if matches := trackUsagePattern.FindStringSubmatch(line); matches != nil {
// P4 labels these fields "us", but reports their values in milliseconds.
track.UserCPU = parseTrackDuration(matches[1], "ms")
track.SystemCPU = parseTrackDuration(matches[2], "ms")
track.tracked = true
continue
}
if matches := trackTablePattern.FindStringSubmatch(line); matches != nil {
table = matches[1]
library = ""
continue
}
if matches := trackLibraryPattern.FindStringSubmatch(line); matches != nil {
library = matches[1]
table = ""
continue
}
if matches := trackLockPattern.FindStringSubmatch(line); matches != nil && table != "" {
track.Tables[table] = tableTrack{
ReadWait: parseTrackDuration(matches[1], "ms"),
ReadHeld: parseTrackDuration(matches[2], "ms"),
WriteWait: parseTrackDuration(matches[3], "ms"),
WriteHeld: parseTrackDuration(matches[4], "ms"),
}
track.tracked = true
continue
}
if matches := trackLibraryUsePattern.FindStringSubmatch(line); matches != nil && library != "" {
track.Libraries[library] = libraryTrack{
Opens: parseTrackCount(matches[1]),
Closes: parseTrackCount(matches[2]),
Checkins: parseTrackCount(matches[3]),
Exists: parseTrackCount(matches[4]),
}
track.tracked = true
}
}
return track
}
func parseTrackCount(value string) int64 {
count, err := strconv.ParseInt(value, 10, 64)
if err != nil {
return 0
}
return count
}
func (track *trackOutput) add(other trackOutput) {
track.Lapse += other.Lapse
track.UserCPU += other.UserCPU
track.SystemCPU += other.SystemCPU
for name, locks := range other.Tables {
total := track.Tables[name]
total.ReadWait += locks.ReadWait
total.ReadHeld += locks.ReadHeld
total.WriteWait += locks.WriteWait
total.WriteHeld += locks.WriteHeld
track.Tables[name] = total
}
for name, usage := range other.Libraries {
total := track.Libraries[name]
total.Opens += usage.Opens
total.Closes += usage.Closes
total.Checkins += usage.Checkins
total.Exists += usage.Exists
track.Libraries[name] = total
}
}
func (collector *trackCollector) add(command string, track trackOutput) {
if collector == nil || !track.tracked {
return
}
collector.mu.Lock()
defer collector.mu.Unlock()
if collector.commands == nil {
collector.commands = make(map[string]trackOutput)
}
total, ok := collector.commands[command]
if !ok {
total = trackOutput{Tables: make(map[string]tableTrack), Libraries: make(map[string]libraryTrack)}
}
total.add(track)
collector.commands[command] = total
}
func p4CommandName(args []string) string {
for _, arg := range args {
switch arg {
case "depots", "dirs", "sizes", "verify", "pull":
return arg
}
}
return "p4"
}
func (collector *trackCollector) writeSummary(output io.Writer, phase string) {
if collector == nil {
return
}
collector.mu.Lock()
defer collector.mu.Unlock()
commands := make([]string, 0, len(collector.commands))
for command := range collector.commands {
commands = append(commands, command)
}
sort.Strings(commands)
for _, command := range commands {
track := collector.commands[command]
fmt.Fprintf(output, "P4 tracking %s %s: lapse=%s user=%s system=%s\n", phase, command, track.Lapse, track.UserCPU, track.SystemCPU)
tables := make([]string, 0, len(track.Tables))
for table := range track.Tables {
tables = append(tables, table)
}
sort.Strings(tables)
for _, table := range tables {
locks := track.Tables[table]
fmt.Fprintf(output, "P4 tracking %s %s %s: read wait/held=%s/%s write wait/held=%s/%s\n", phase, command, table, locks.ReadWait, locks.ReadHeld, locks.WriteWait, locks.WriteHeld)
}
libraries := make([]string, 0, len(track.Libraries))
for library := range track.Libraries {
libraries = append(libraries, library)
}
sort.Strings(libraries)
for _, library := range libraries {
usage := track.Libraries[library]
fmt.Fprintf(output, "P4 tracking %s %s lbr %s: opens/closes/checkins/exists=%d/%d/%d/%d\n", phase, command, library, usage.Opens, usage.Closes, usage.Checkins, usage.Exists)
}
}
}
func (p p4Client) commandArgs(args ...string) []string {
commandArgs := make([]string, 0, len(args)+5)
commandArgs = append(commandArgs, "-Ztrack")
if p.port != "" {
commandArgs = append(commandArgs, "-p", p.port)
}
if p.user != "" {
commandArgs = append(commandArgs, "-u", p.user)
}
return append(commandArgs, args...)
}
func (p p4Client) run(ctx context.Context, args ...string) ([]byte, error) {
commandArgs := p.commandArgs(args...)
p.logger.Debugf("Running: %s %s", p.bin, strings.Join(commandArgs, " "))
output, err := exec.CommandContext(ctx, p.bin, commandArgs...).CombinedOutput()
p.recordTrack(args, string(output))
if err != nil {
return output, fmt.Errorf("p4 %s: %w: %s", strings.Join(args, " "), err, strings.TrimSpace(string(output)))
}
return output, nil
}
func (p p4Client) recordTrack(args []string, output string) {
track := parseTrackOutput(output)
if !track.tracked {
return
}
command := p4CommandName(args)
p.tracking.add(command, track)
p.logger.Debugf("P4 tracking %s: lapse=%s user=%s system=%s", command, track.Lapse, track.UserCPU, track.SystemCPU)
for table, locks := range track.Tables {
p.logger.Debugf("P4 tracking %s %s: read wait/held=%s/%s write wait/held=%s/%s", command, table, locks.ReadWait, locks.ReadHeld, locks.WriteWait, locks.WriteHeld)
}
for library, usage := range track.Libraries {
p.logger.Debugf("P4 tracking %s lbr %s: opens/closes/checkins/exists=%d/%d/%d/%d", command, library, usage.Opens, usage.Closes, usage.Checkins, usage.Exists)
}
}
func missingRevision(line string) (string, bool) {
matches := missingRevisionPattern.FindStringSubmatch(line)
if matches == nil {
return "", false
}
return matches[1], true
}
func (p p4Client) discoverMissing(ctx context.Context, filespec string, output io.Writer) (int64, error) {
filespecs := strings.Fields(filespec)
if len(filespecs) == 0 {
return 0, errors.New("verify requires at least one depot path")
}
args := append([]string{"verify", "-q", "--only", "MISSING"}, filespecs...)
commandArgs := p.commandArgs(args...)
p.logger.Debugf("Running: %s %s", p.bin, strings.Join(commandArgs, " "))
command := exec.CommandContext(ctx, p.bin, commandArgs...)
stdout, err := command.StdoutPipe()
if err != nil {
return 0, err
}
stderr, err := command.StderrPipe()
if err != nil {
return 0, err
}
if err := command.Start(); err != nil {
return 0, err
}
type scanResult struct {
found int64
diagnostics string
tracking string
err error
}
results := make(chan scanResult, 2)
var outputMu sync.Mutex
scan := func(reader io.Reader) {
result := scanResult{}
var diagnostics strings.Builder
var tracking strings.Builder
scanner := bufio.NewScanner(reader)
scanner.Buffer(make([]byte, 64*1024), 4*1024*1024)
for scanner.Scan() {
line := scanner.Text()
if strings.HasPrefix(line, "--- ") {
tracking.WriteString(line)
tracking.WriteByte('\n')
continue
}
if revision, ok := missingRevision(line); ok {
if result.err == nil {
outputMu.Lock()
_, result.err = fmt.Fprintln(output, revision)
outputMu.Unlock()
}
result.found++
continue
}
if strings.TrimSpace(line) != "" {
diagnostics.WriteString(line)
diagnostics.WriteByte('\n')
}
}
if err := scanner.Err(); err != nil && result.err == nil {
result.err = err
}
result.diagnostics = diagnostics.String()
result.tracking = tracking.String()
results <- result
}
go scan(stdout)
go scan(stderr)
first, second := <-results, <-results
found := first.found + second.found
waitErr := command.Wait()
p.recordTrack(args, first.tracking+second.tracking)
if first.err != nil {
return found, first.err
}
if second.err != nil {
return found, second.err
}
if waitErr != nil {
var exitError *exec.ExitError
if errors.As(waitErr, &exitError) && exitError.ExitCode() == 1 && found > 0 {
return found, nil
}
diagnostics := strings.TrimSpace(first.diagnostics + second.diagnostics)
return found, fmt.Errorf("p4 %s: %w: %s", strings.Join(args, " "), waitErr, diagnostics)
}
return found, nil
}
func (p p4Client) localOrStreamDepots(ctx context.Context) ([]depot, error) {
output, err := p.run(ctx, "-ztag", "depots")
if err != nil {
return nil, err
}
var depots []depot
current := depot{}
addDepot := func() {
if (current.typeName == "local" || current.typeName == "stream") && current.name != "" {
depots = append(depots, current)
}
}
for _, line := range strings.Split(string(output), "\n") {
fields := strings.Fields(line)
if len(fields) < 3 || fields[0] != "..." {
continue
}
key := strings.ToLower(fields[1])
value := strings.Join(fields[2:], " ")
switch key {
case "depot", "name":
addDepot()
current = depot{name: value}
case "type":
current.typeName = strings.ToLower(value)
case "streamdepth", "depth":
depth, err := strconv.Atoi(value)
if err != nil || depth < 1 {
return nil, fmt.Errorf("invalid stream depth %q for depot %q", value, current.name)
}
current.streamDepth = depth
}
}
addDepot()
return depots, nil
}
func (p p4Client) directories(ctx context.Context, filespec string) ([]string, error) {
output, err := p.run(ctx, "dirs", "-D", filespec)
if err != nil {
return nil, err
}
var directories []string
for _, line := range strings.Split(string(output), "\n") {
directory := strings.TrimSpace(line)
if strings.Contains(directory, " - no such file(s).") {
p.logger.Warnf("Ignoring p4 dirs diagnostic: %s", directory)
continue
}
if strings.HasPrefix(directory, "//") {
directories = append(directories, directory)
}
}
return directories, nil
}
func (p p4Client) streamDirectories(ctx context.Context, name string, depth int) ([]string, error) {
if depth < 1 {
return nil, nil
}
pattern := "//" + name + "/" + strings.TrimSuffix(strings.Repeat("*/", depth), "/")
output, err := p.run(ctx, "dirs", pattern)
if err != nil {
return nil, err
}
var directories []string
for _, line := range strings.Split(string(output), "\n") {
directory := strings.TrimSpace(line)
if strings.Contains(directory, " - no such file(s).") {
p.logger.Warnf("Ignoring p4 dirs diagnostic: %s", directory)
continue
}
if strings.HasPrefix(directory, "//"+name+"/") {
directories = append(directories, directory)
}
}
return directories, nil
}
func (p p4Client) sizesWithinTimeout(ctx context.Context, filespec string, timeout time.Duration) (bool, error) {
timedContext, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
_, err := p.run(timedContext, "sizes", "-sh", filespec)
if err == nil {
return true, nil
}
if errors.Is(timedContext.Err(), context.DeadlineExceeded) {
return false, nil
}
return false, err
}
func (p p4Client) filesWithinTimeout(ctx context.Context, filespec string, timeout time.Duration) (int64, error) {
timedContext, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
output, err := p.run(timedContext, "sizes", "-sh", filespec)
if err != nil {
if errors.Is(timedContext.Err(), context.DeadlineExceeded) {
return 0, fmt.Errorf("p4 sizes timed out for %q", filespec)
}
return 0, err
}
matches := sizesFileCountPattern.FindStringSubmatch(string(output))
if matches == nil {
return 0, fmt.Errorf("could not parse file count from p4 sizes output for %q", filespec)
}
files, err := strconv.ParseInt(strings.ReplaceAll(matches[1], ",", ""), 10, 64)
if err != nil {
return 0, fmt.Errorf("parse p4 sizes file count for %q: %w", filespec, err)
}
return files, nil
}
func groupStreamPaths(name string, directories []string, depth int, averageFiles, maxFiles int64) ([]string, error) {
if len(directories) == 0 {
return nil, nil
}
prefix := "//" + name + "/"
partsByDirectory := make(map[string][]string, len(directories))
for _, directory := range directories {
relative := strings.TrimPrefix(directory, prefix)
parts := strings.Split(strings.Trim(relative, "/"), "/")
if relative == directory || len(parts) != depth {
return nil, fmt.Errorf("stream directory %q does not match depth %d", directory, depth)
}
partsByDirectory[directory] = parts
}
type streamGroup struct {
filespec string
streams int
}
var grouped []streamGroup
var split func([]string, int)
split = func(candidates []string, level int) {
groups := make(map[string][]string)
for _, candidate := range candidates {
key := strings.Join(partsByDirectory[candidate][:level], "/")
groups[key] = append(groups[key], candidate)
}
keys := make([]string, 0, len(groups))
for key := range groups {
keys = append(keys, key)
}
sort.Strings(keys)
for _, key := range keys {
candidates := groups[key]
if int64(len(candidates))*averageFiles <= maxFiles || level == depth {
grouped = append(grouped, streamGroup{filespec: prefix + key + "/...", streams: len(candidates)})
continue
}
split(candidates, level+1)
}
}
split(directories, 1)
var batches []string
var batch []string
var batchFiles int64
flushBatch := func() {
if len(batch) == 0 {
return
}
batches = append(batches, strings.Join(batch, " "))
batch = nil
batchFiles = 0
}
for _, group := range grouped {
groupFiles := int64(group.streams) * averageFiles
if len(batch) > 0 && batchFiles+groupFiles > maxFiles {
flushBatch()
}
batch = append(batch, group.filespec)
batchFiles += groupFiles
}
flushBatch()
return batches, nil
}
func (p p4Client) generateStreamPaths(ctx context.Context, stream depot, timeout time.Duration, maxFiles int64, samples int) ([]string, error) {
directories, err := p.streamDirectories(ctx, stream.name, stream.streamDepth)
if err != nil {
return nil, err
}
if len(directories) == 0 {
return nil, nil
}
if samples > len(directories) {
samples = len(directories)
}
random := rand.New(rand.NewSource(time.Now().UnixNano()))
indices := random.Perm(len(directories))[:samples]
var totalFiles int64
for _, index := range indices {
files, err := p.filesWithinTimeout(ctx, strings.TrimSuffix(directories[index], "/")+"/...", timeout)
if err != nil {
return nil, err
}
totalFiles += files
}
averageFiles := (totalFiles + int64(samples) - 1) / int64(samples)
return groupStreamPaths(stream.name, directories, stream.streamDepth, averageFiles, maxFiles)
}
func generateDepotPaths(ctx context.Context, p4 p4Client, output string, timeout time.Duration, maxStreamFiles int64, streamSizeSamples int) (int, error) {
return generateDepotPathsWithWorkers(ctx, p4, output, timeout, maxStreamFiles, streamSizeSamples, 10)
}
func (p p4Client) generateLocalDepotPaths(ctx context.Context, name string, timeout time.Duration) ([]string, error) {
paths := []string{"//" + name + "/..."}
var generated []string
for len(paths) > 0 {
path := paths[0]
paths = paths[1:]
complete, err := p.sizesWithinTimeout(ctx, path, timeout)
if err != nil {
return nil, err
}
if complete {
generated = append(generated, path)
continue
}
directories, err := p.directories(ctx, strings.TrimSuffix(path, "/...")+"/*")
if err != nil {
return nil, err
}
if len(directories) == 0 {
return nil, fmt.Errorf("cannot subdivide timed-out depot path %q", path)
}
for _, directory := range directories {
paths = append(paths, strings.TrimSuffix(directory, "/")+"/...")
}
}
return generated, nil
}
func generateDepotPathsWithWorkers(ctx context.Context, p4 p4Client, output string, timeout time.Duration, maxStreamFiles int64, streamSizeSamples, workers int) (int, error) {
if timeout <= 0 {
return 0, errors.New("path timeout must be positive")
}
if maxStreamFiles < 1 || streamSizeSamples < 1 || workers < 1 {
return 0, errors.New("maximum stream files, stream size samples, and path workers must be positive")
}
if _, err := os.Stat(output); err == nil {
return 0, fmt.Errorf("depot path output %q already exists", output)
} else if !errors.Is(err, os.ErrNotExist) {
return 0, err
}
depots, err := p4.localOrStreamDepots(ctx)
if err != nil {
return 0, err
}
if workers > len(depots) {
workers = len(depots)
}
generatedByDepot := make([][]string, len(depots))
workerContext, cancel := context.WithCancel(ctx)
defer cancel()
jobs := make(chan int)
errorsCh := make(chan error, 1)
var workersDone sync.WaitGroup
for range workers {
workersDone.Add(1)
go func() {
defer workersDone.Done()
for index := range jobs {
depot := depots[index]
var paths []string
var err error
if depot.typeName == "stream" {
paths, err = p4.generateStreamPaths(workerContext, depot, timeout, maxStreamFiles, streamSizeSamples)
} else {
paths, err = p4.generateLocalDepotPaths(workerContext, depot.name, timeout)
}
if err != nil {
select {
case errorsCh <- err:
cancel()
default:
}
return
}
generatedByDepot[index] = paths
}
}()
}
dispatch:
for index := range depots {
select {
case jobs <- index:
case <-workerContext.Done():
break dispatch
}
}
close(jobs)
workersDone.Wait()
select {
case err := <-errorsCh:
return 0, err
default:
}
if ctx.Err() != nil {
return 0, ctx.Err()
}
var generated []string
for _, paths := range generatedByDepot {
generated = append(generated, paths...)
}
outputDir := filepath.Dir(output)
if err := os.MkdirAll(outputDir, 0755); err != nil && outputDir != "." {
return 0, err
}
temporary, err := os.CreateTemp(outputDir, ".seedreplica-paths-*.txt")
if err != nil {
return 0, err
}
temporaryName := temporary.Name()
defer os.Remove(temporaryName)
for _, path := range generated {
if _, err := fmt.Fprintln(temporary, path); err != nil {
temporary.Close()
return 0, err
}
}
if err := temporary.Close(); err != nil {
return 0, err
}
if err := os.Rename(temporaryName, output); err != nil {
return 0, err
}
return len(generated), nil
}
func discoverRevisions(ctx context.Context, p4 p4Client, filespecs, output string, workers int, replace bool) (int64, error) {
if workers < 1 {
return 0, errors.New("discovery workers must be positive")
}
if _, err := os.Stat(output); err == nil && !replace {
return 0, fmt.Errorf("discovery output %q already exists; use --replace-input to overwrite it", output)
} else if err != nil && !errors.Is(err, os.ErrNotExist) {
return 0, err
}
input, err := os.Open(filespecs)
if err != nil {
return 0, err
}
defer input.Close()
outputDir := filepath.Dir(output)
if err := os.MkdirAll(outputDir, 0755); err != nil && outputDir != "." {
return 0, err
}
workerFiles := make([]*os.File, workers)
for index := range workerFiles {
workerFiles[index], err = os.CreateTemp(outputDir, ".seedreplica-discovery-*.txt")
if err != nil {
for _, file := range workerFiles {
if file != nil {
file.Close()
os.Remove(file.Name())
}
}
return 0, err
}
defer os.Remove(workerFiles[index].Name())
}
workerContext, cancel := context.WithCancel(ctx)
defer cancel()
jobs := make(chan string)
errorsCh := make(chan error, 1)
counts := make([]int64, workers)
var workersDone sync.WaitGroup
for index, file := range workerFiles {
workersDone.Add(1)
go func(index int, file *os.File) {
defer workersDone.Done()
writer := bufio.NewWriter(file)
defer writer.Flush()
for {
select {
case <-workerContext.Done():
return
case filespec, ok := <-jobs:
if !ok {
return
}
count, err := p4.discoverMissing(workerContext, filespec, writer)
if err != nil {
select {
case errorsCh <- err:
cancel()
default:
}
return
}
counts[index] += count
}
}
}(index, file)
}
scanner := bufio.NewScanner(input)
scanner.Buffer(make([]byte, 64*1024), 4*1024*1024)
var producerErr error
for scanner.Scan() {
filespecs := strings.Fields(scanner.Text())
if len(filespecs) == 0 {
continue
}
for _, filespec := range filespecs {
if !strings.HasPrefix(filespec, "//") {
producerErr = fmt.Errorf("discovery input contains an invalid depot path: %q", filespec)
cancel()
break
}
}
if producerErr != nil {
break
}
select {
case jobs <- strings.Join(filespecs, " "):
case <-workerContext.Done():
}
if workerContext.Err() != nil {
break
}
}
scanErr := scanner.Err()
close(jobs)
workersDone.Wait()
for _, file := range workerFiles {
if err := file.Close(); err != nil {
return 0, err
}
}
if scanErr != nil {
return 0, scanErr
}
if producerErr != nil {
return 0, producerErr
}
select {
case err := <-errorsCh:
return 0, err
default:
}
if ctx.Err() != nil {
return 0, ctx.Err()
}
temporary, err := os.CreateTemp(outputDir, ".seedreplica-input-*.txt")
if err != nil {
return 0, err
}
temporaryName := temporary.Name()
defer os.Remove(temporaryName)
var total int64
for index, file := range workerFiles {
contents, err := os.ReadFile(file.Name())
if err != nil {
temporary.Close()
return 0, err
}
if _, err := temporary.Write(contents); err != nil {
temporary.Close()
return 0, err
}
total += counts[index]
}
if err := temporary.Close(); err != nil {
return 0, err
}
if replace {
if err := os.Remove(output); err != nil && !errors.Is(err, os.ErrNotExist) {
return 0, err
}
}
if err := os.Rename(temporaryName, output); err != nil {
return 0, err
}
return total, nil
}
func (p p4Client) queueStatus(ctx context.Context) (queueStatus, error) {
output, err := p.run(ctx, "pull", "-lsh")
if err != nil {
return queueStatus{}, err
}
return parseQueueStatus(string(output))
}
func parseQueueStatus(output string) (queueStatus, error) {
matches := pullQueuePattern.FindStringSubmatch(output)
if matches == nil {
return queueStatus{}, fmt.Errorf("could not parse p4 pull -lsh output: %q", strings.TrimSpace(output))
}
active, activeErr := strconv.Atoi(matches[1])
total, totalErr := strconv.Atoi(matches[2])
if activeErr != nil || totalErr != nil {
return queueStatus{}, errors.New("invalid transfer counts in p4 pull -lsh output")
}
return queueStatus{ActiveTransfers: active, TotalTransfers: total, ActiveBytes: matches[3], TotalBytes: matches[4]}, nil
}
func readBatch(filename string, offset, linesRead int64, batchSize int) ([]string, int64, int64, bool, error) {
input, err := os.Open(filename)
if err != nil {
return nil, offset, linesRead, false, err
}
defer input.Close()
if _, err := input.Seek(offset, io.SeekStart); err != nil {
return nil, offset, linesRead, false, err
}
reader := bufio.NewReader(input)
batch := make([]string, 0, batchSize)
for len(batch) < batchSize {
line, readErr := reader.ReadString('\n')
if len(line) > 0 {
offset += int64(len(line))
linesRead++
value := strings.TrimSpace(line)
if value != "" {
if !strings.HasPrefix(value, "//") || !strings.Contains(value, "#") {
return nil, offset, linesRead, false, fmt.Errorf("input line %d is not a depot revision: %q", linesRead, value)
}
batch = append(batch, value)
}
}
if readErr == io.EOF {
return batch, offset, linesRead, true, nil
}
if readErr != nil {
return nil, offset, linesRead, false, readErr
}
}
return batch, offset, linesRead, false, nil
}
func writeCheckpoint(filename string, state checkpoint) error {
state.UpdatedAt = time.Now().UTC()
encoded, err := json.MarshalIndent(state, "", " ")
if err != nil {
return err
}
if err := os.MkdirAll(filepath.Dir(filename), 0755); err != nil && filepath.Dir(filename) != "." {
return err
}
temporary, err := os.CreateTemp(filepath.Dir(filename), ".seedreplica-*.json")
if err != nil {
return err
}
temporaryName := temporary.Name()
defer os.Remove(temporaryName)
if _, err := temporary.Write(append(encoded, '\n')); err != nil {
temporary.Close()
return err
}
if err := temporary.Close(); err != nil {
return err
}
return os.Rename(temporaryName, filename)
}
func loadCheckpoint(filename, input string) (checkpoint, error) {
encoded, err := os.ReadFile(filename)
if err != nil {
return checkpoint{}, err
}
var state checkpoint
if err := json.Unmarshal(encoded, &state); err != nil {
return checkpoint{}, fmt.Errorf("decode checkpoint: %w", err)
}
if state.Input != input {
return checkpoint{}, fmt.Errorf("checkpoint input %q does not match %q", state.Input, input)
}
return state, nil
}
func verifyBatch(ctx context.Context, p4 p4Client, batch []string) (int, error) {
temporary, err := os.CreateTemp("", "seedreplica-batch-*.txt")
if err != nil {
return 0, err
}
temporaryName := temporary.Name()
defer os.Remove(temporaryName)
if _, err := temporary.WriteString(strings.Join(batch, "\n") + "\n"); err != nil {
temporary.Close()
return 0, err
}
if err := temporary.Close(); err != nil {
return 0, err
}
output, err := p4.run(ctx, "-x", temporaryName, "verify", "-qt", "--only", "MISSING")
if err != nil {
return 0, err
}
transfers := 0
for _, line := range strings.Split(string(output), "\n") {
if strings.Contains(line, "BAD") || strings.Contains(line, "MISSING") {
transfers++
}
}
return transfers, nil
}
func waitForQueue(ctx context.Context, p4 p4Client, maxQueue int, interval time.Duration) error {
for {
status, err := p4.queueStatus(ctx)
if err != nil {
return err
}
p4.logger.Infof("Pull queue: %d active/%d total transfers, %s active/%s total bytes", status.ActiveTransfers, status.TotalTransfers, status.ActiveBytes, status.TotalBytes)
if status.TotalTransfers <= maxQueue {
return nil
}
p4.logger.Infof("Pull queue size %d exceeds threshold %d; waiting %s", status.TotalTransfers, maxQueue, interval)
timer := time.NewTimer(interval)
select {
case <-ctx.Done():
timer.Stop()
return ctx.Err()
case <-timer.C:
}
}
}
func runSeed(ctx context.Context, cfg config, logger *logrus.Logger, stdout io.Writer) error {
tracking := &trackCollector{}
defer tracking.writeSummary(stdout, "seed")
input, err := filepath.Abs(cfg.input)
if err != nil {
return err
}
if cfg.discoverInput != "" {
filespecs, err := filepath.Abs(cfg.discoverInput)
if err != nil {
return err
}
p4 := p4Client{bin: cfg.p4bin, port: cfg.p4port, user: cfg.p4user, logger: logger, tracking: tracking}
logger.Infof("Discovering missing revisions from %s with %d workers", filespecs, cfg.discoverers)
found, err := discoverRevisions(ctx, p4, filespecs, input, cfg.discoverers, cfg.replaceInput)
if err != nil {
return fmt.Errorf("discover missing revisions: %w", err)
}
logger.Infof("Discovery wrote %d missing revisions to %s", found, input)
if cfg.discoverOnly {
return nil
}
}
state := checkpoint{Input: input}
if cfg.resume {
state, err = loadCheckpoint(cfg.checkpoint, input)
if err != nil {
return err
}
logger.Infof("Resuming after %d processed revisions in %d batches", state.Processed, state.Batches)
}
p4 := p4Client{bin: cfg.p4bin, port: cfg.p4port, user: cfg.p4user, logger: logger, tracking: tracking}
for {
batch, offset, linesRead, eof, err := readBatch(input, state.Offset, state.LinesRead, cfg.batchSize)
if err != nil {
return err
}
if len(batch) == 0 {
if !cfg.follow || !eof {
state.CompletedAt = time.Now().UTC()
return writeCheckpoint(cfg.checkpoint, state)
}
logger.Infof("Waiting for additional input in %s", input)
timer := time.NewTimer(cfg.pollInterval)
select {
case <-ctx.Done():
timer.Stop()
return ctx.Err()
case <-timer.C:
}
continue
}
batchNumber := state.Batches + 1
transfers := 0
if cfg.dryRun {
logger.Infof("Dry run: batch %d contains %d revisions", batchNumber, len(batch))
} else {
transfers, err = verifyBatch(ctx, p4, batch)
if err != nil {
return fmt.Errorf("batch %d: %w", batchNumber, err)
}
}
state.Offset = offset
state.LinesRead = linesRead
state.Processed += int64(len(batch))
state.Batches++
if err := writeCheckpoint(cfg.checkpoint, state); err != nil {
return err
}
logger.Infof("Batch %d queued %d transfers from %d revisions (%d total processed)", batchNumber, transfers, len(batch), state.Processed)
if !cfg.dryRun {
if err := waitForQueue(ctx, p4, cfg.maxQueue, cfg.pollInterval); err != nil {
return err
}
}
if cfg.maxBatches > 0 && state.Batches >= cfg.maxBatches {
logger.Infof("Reached maximum batch limit %d", cfg.maxBatches)
return nil
}
}
}
func commandApp(name string, stderr io.Writer) *kingpin.Application {
app := kingpin.New(name, description)
app.ErrorWriter(stderr)
app.UsageWriter(stderr)
app.HelpFlag.Short('h')
app.Version(VersionPrint(name, "v"+Version))
app.VersionFlag.Short('V')
return app
}
func commandLogger(stderr io.Writer, debug bool) *logrus.Logger {
logger := logrus.New()
logger.SetOutput(stderr)
if debug {
logger.SetLevel(logrus.DebugLevel)
}
return logger
}
func runPaths(args []string, stdout, stderr io.Writer) error {
app := commandApp("p4seedreplica discover", stderr)
output := app.Flag("output", "File to receive generated depot paths").Short('o').Required().String()
p4bin := app.Flag("p4bin", "Path to p4 executable").Default("p4").String()
p4port := app.Flag("p4port", "P4PORT override").Short('p').String()
p4user := app.Flag("p4user", "P4USER override").Short('u').String()
pathTimeout := app.Flag("path-timeout", "Maximum duration for each p4 sizes command").Default("5m").Duration()
maxStreamFiles := app.Flag("max-stream-files", "Estimated maximum files in each generated stream path").Default("1000000").Int64()
streamSizeSamples := app.Flag("stream-size-samples", "Random stream paths sampled to estimate file counts").Default("10").Int()
pathWorkers := app.Flag("path-workers", "Parallel depot path generation workers").Default("10").Int()
debug := app.Flag("debug", "Print p4 commands").Bool()
if _, err := app.Parse(args); err != nil {
return err
}
if *pathTimeout <= 0 || *maxStreamFiles < 1 || *streamSizeSamples < 1 || *pathWorkers < 1 {
return errors.New("path timeout, maximum stream files, stream size samples, and path workers must be positive")
}
logger := commandLogger(stderr, *debug)
tracking := &trackCollector{}
defer tracking.writeSummary(stdout, "discover")
count, err := generateDepotPathsWithWorkers(context.Background(), p4Client{bin: *p4bin, port: *p4port, user: *p4user, logger: logger, tracking: tracking}, *output, *pathTimeout, *maxStreamFiles, *streamSizeSamples, *pathWorkers)
if err != nil {
return fmt.Errorf("generate depot paths: %w", err)
}
fmt.Fprintf(stdout, "Wrote %d depot paths to %s\n", count, *output)
return nil
}
func runVerify(args []string, stdout, stderr io.Writer) error {
app := commandApp("p4seedreplica verify", stderr)
paths := app.Flag("paths", "File of depot path specs to scan").Required().String()
output := app.Flag("output", "File to receive missing depot revisions").Short('o').Required().String()
p4bin := app.Flag("p4bin", "Path to p4 executable").Default("p4").String()
p4port := app.Flag("p4port", "P4PORT override").Short('p').String()
p4user := app.Flag("p4user", "P4USER override").Short('u').String()
workers := app.Flag("workers", "Parallel missing-file discovery workers").Default("10").Int()
replace := app.Flag("replace", "Allow replacing an existing output file").Bool()
debug := app.Flag("debug", "Print p4 commands").Bool()
if _, err := app.Parse(args); err != nil {
return err
}
if *workers < 1 {
return errors.New("discovery workers must be positive")
}
logger := commandLogger(stderr, *debug)
tracking := &trackCollector{}
defer tracking.writeSummary(stdout, "verify")
pathsFile, err := filepath.Abs(*paths)
if err != nil {
return err
}
outputFile, err := filepath.Abs(*output)
if err != nil {
return err
}
logger.Infof("Verifying depot paths from %s with %d workers", pathsFile, *workers)
found, err := discoverRevisions(context.Background(), p4Client{bin: *p4bin, port: *p4port, user: *p4user, logger: logger, tracking: tracking}, pathsFile, outputFile, *workers, *replace)
if err != nil {
return fmt.Errorf("verify missing revisions: %w", err)
}
fmt.Fprintf(stdout, "Wrote %d missing revisions to %s\n", found, outputFile)
return nil
}
func runSeedCommand(args []string, stdout, stderr io.Writer) error {
app := commandApp("p4seedreplica seed", stderr)
input := app.Flag("input", "File of depot revisions, one //depot/file#rev per line").Required().String()
p4bin := app.Flag("p4bin", "Path to p4 executable").Default("p4").String()
p4port := app.Flag("p4port", "P4PORT override").Short('p').String()
p4user := app.Flag("p4user", "P4USER override").Short('u').String()
batchSize := app.Flag("batch-size", "Revisions submitted per verify command").Default("10000").Int()
maxQueue := app.Flag("max-pull-queue", "Maximum queued archive transfers before waiting").Default("100000").Int()
pollInterval := app.Flag("poll-interval", "Queue polling interval").Default("30s").Duration()
maxBatches := app.Flag("max-batches", "Stop after this many batches; zero means unlimited").Default("0").Int()
checkpointPath := app.Flag("checkpoint", "Checkpoint file (default: input.seedreplica.json)").String()
resume := app.Flag("resume", "Resume from checkpoint").Bool()
follow := app.Flag("follow", "Wait for appended input after reaching end of file").Bool()
dryRun := app.Flag("dry-run", "Validate and checkpoint batches without running p4").Bool()
debug := app.Flag("debug", "Print p4 commands").Bool()
if _, err := app.Parse(args); err != nil {
return err
}
if *batchSize < 1 || *maxQueue < 1 || *pollInterval <= 0 || *maxBatches < 0 {
return errors.New("batch size, maximum queue, and polling interval must be positive; maximum batches cannot be negative")
}
checkpoint := *checkpointPath
if checkpoint == "" {
checkpoint = *input + ".seedreplica.json"
}
return runSeed(context.Background(), config{
input: *input, p4bin: *p4bin, p4port: *p4port, p4user: *p4user, batchSize: *batchSize,
maxQueue: *maxQueue, pollInterval: *pollInterval, maxBatches: *maxBatches, checkpoint: checkpoint,
resume: *resume, follow: *follow, dryRun: *dryRun,
}, commandLogger(stderr, *debug), stdout)
}
func workflowUsage(stderr io.Writer) {
fmt.Fprintln(stderr, `usage: p4seedreplica <command> [<flags>]
Commands reflect the 3 phases of the process:
discover Generate bounded depot paths for verification.
verify Find missing archive revisions from depot paths using verify command.
seed Submit missing revisions to the replica pull queue.
Run "p4seedreplica <command> --help" for command options.`)
}
func run(args []string, stdout, stderr io.Writer) error {
if len(args) == 0 || args[0] == "help" || args[0] == "--help" || args[0] == "-h" {
workflowUsage(stderr)
return nil
}
switch args[0] {
case "discover":
return runPaths(args[1:], stdout, stderr)
case "verify":
return runVerify(args[1:], stdout, stderr)
case "seed":
return runSeedCommand(args[1:], stdout, stderr)
default:
return runLegacy(args, stdout, stderr)
}
}
func runLegacy(args []string, stdout, stderr io.Writer) error {
app := commandApp("p4seedreplica", stderr)
input := app.Flag("input", "File of depot revisions, one //depot/file#rev per line").String()
p4bin := app.Flag("p4bin", "Path to p4 executable").Default("p4").String()
p4port := app.Flag("p4port", "P4PORT override").Short('p').String()
p4user := app.Flag("p4user", "P4USER override").Short('u').String()
batchSize := app.Flag("batch-size", "Revisions submitted per verify command").Default("10000").Int()
maxQueue := app.Flag("max-pull-queue", "Maximum queued archive transfers before waiting").Default("100000").Int()
pollInterval := app.Flag("poll-interval", "Queue and follow polling interval").Default("30s").Duration()
maxBatches := app.Flag("max-batches", "Stop after this many batches; zero means unlimited").Default("0").Int()
checkpointPath := app.Flag("checkpoint", "Checkpoint file (default: input.seedreplica.json)").String()
resume := app.Flag("resume", "Resume from checkpoint").Bool()
follow := app.Flag("follow", "Wait for appended input after reaching end of file").Bool()
dryRun := app.Flag("dry-run", "Validate and checkpoint batches without running p4").Bool()
discoverInput := app.Flag("discover-input", "File of depot path specs to scan in parallel for missing revisions").String()
discoverOnly := app.Flag("discover-only", "Stop after writing revisions discovered from --discover-input").Bool()
discoverWorkers := app.Flag("discover-workers", "Parallel missing-file discovery workers").Default("10").Int()
replaceInput := app.Flag("replace-input", "Allow discovery to replace an existing --input file").Bool()
depotPaths := app.Flag("generate-depot-paths", "Write recursively sized local and stream depot paths to this file").String()
pathTimeout := app.Flag("path-timeout", "Maximum duration for each p4 sizes command while generating depot paths").Default("5m").Duration()
maxStreamFiles := app.Flag("max-stream-files", "Estimated maximum files in each generated stream path").Default("1000000").Int64()
streamSizeSamples := app.Flag("stream-size-samples", "Random stream paths sampled to estimate file counts").Default("10").Int()
pathWorkers := app.Flag("path-workers", "Parallel depot path generation workers").Default("10").Int()
debug := app.Flag("debug", "Print p4 commands").Bool()
if _, err := app.Parse(args); err != nil {
return err
}
if *batchSize < 1 || *maxQueue < 1 || *pollInterval <= 0 || *maxBatches < 0 || *discoverWorkers < 1 || *pathTimeout <= 0 || *maxStreamFiles < 1 || *streamSizeSamples < 1 || *pathWorkers < 1 {
return errors.New("batch size, maximum queue, and polling interval must be positive; maximum batches cannot be negative")
}
if *input == "" && *depotPaths == "" {
return errors.New("--input is required unless --generate-depot-paths is used")
}
if *depotPaths != "" {
if *discoverInput != "" || *discoverOnly || *resume || *follow || *dryRun {
return errors.New("--generate-depot-paths cannot be combined with discovery or seeding options")
}
logger := logrus.New()
logger.SetOutput(stderr)
if *debug {
logger.SetLevel(logrus.DebugLevel)
}
tracking := &trackCollector{}
defer tracking.writeSummary(stdout, "discover")
count, err := generateDepotPathsWithWorkers(context.Background(), p4Client{bin: *p4bin, port: *p4port, user: *p4user, logger: logger, tracking: tracking}, *depotPaths, *pathTimeout, *maxStreamFiles, *streamSizeSamples, *pathWorkers)
if err != nil {
return fmt.Errorf("generate depot paths: %w", err)
}
fmt.Fprintf(stdout, "Wrote %d depot paths to %s\n", count, *depotPaths)
return nil
}
if *discoverOnly && *discoverInput == "" {
return errors.New("--discover-only requires --discover-input")
}
if *dryRun && *discoverInput != "" {
return errors.New("--dry-run cannot be combined with --discover-input")
}
if *resume && *discoverInput != "" {
return errors.New("--resume cannot be combined with --discover-input")
}
checkpoint := *checkpointPath
if checkpoint == "" {
checkpoint = *input + ".seedreplica.json"
}
logger := logrus.New()
logger.SetOutput(stderr)
if *debug {
logger.SetLevel(logrus.DebugLevel)
}
return runSeed(context.Background(), config{
input: *input, p4bin: *p4bin, p4port: *p4port, p4user: *p4user, batchSize: *batchSize,
maxQueue: *maxQueue, pollInterval: *pollInterval, maxBatches: *maxBatches, checkpoint: checkpoint,
resume: *resume, follow: *follow, dryRun: *dryRun, discoverInput: *discoverInput,
discoverOnly: *discoverOnly, discoverers: *discoverWorkers, replaceInput: *replaceInput,
}, logger, stdout)
}
func main() {
if err := run(os.Args[1:], os.Stdout, os.Stderr); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
}