package main import ( "context" "errors" "fmt" "github.com/c2h5oh/datasize" "github.com/sethvargo/go-retry" log "github.com/sirupsen/logrus" "os" "os/exec" "path/filepath" "somegit.dev/ALHP/ALHP.GO/ent/dbpackage" "strings" "sync" "time" ) const MaxUnknownBuilder = 2 type BuildManager struct { repoPurge map[string]chan []*ProtoPackage repoAdd map[string]chan []*ProtoPackage repoWG *sync.WaitGroup alpmMutex *sync.RWMutex building []*ProtoPackage buildingLock *sync.RWMutex queueSignal chan struct{} } func (b *BuildManager) buildQueue(ctx context.Context, queue []*ProtoPackage) error { var ( doneQ []*ProtoPackage doneQLock = new(sync.RWMutex) unknownBuilds bool queueNoMatch bool ) for len(doneQ) != len(queue) { up := 0 b.buildingLock.RLock() if (pkgList2MaxMem(b.building) < conf.Build.MemoryLimit && !unknownBuilds && !queueNoMatch) || (unknownBuilds && len(b.building) < MaxUnknownBuilder) { queueNoMatch = true b.buildingLock.RUnlock() for _, pkg := range queue { // check if package is already build doneQLock.RLock() if ContainsPkg(doneQ, pkg, true) { doneQLock.RUnlock() continue } doneQLock.RUnlock() // check if package is already building (we do not build packages from different marchs simultaneously) b.buildingLock.RLock() if ContainsPkg(b.building, pkg, false) { log.Debugf("[Q] skipped already building package %s->%s", pkg.FullRepo, pkg.Pkgbase) b.buildingLock.RUnlock() continue } b.buildingLock.RUnlock() // only check for memory on known-memory-builds // otherwise build them one-at-a-time // TODO: add initial compile mode for new repos if !unknownBuilds { // check if package has unknown memory usage if pkg.DBPackage.MaxRss == nil { log.Debugf("[Q] skipped unknown package %s->%s", pkg.FullRepo, pkg.Pkgbase) up++ continue } // check if package can be built with current memory limit if datasize.ByteSize(*pkg.DBPackage.MaxRss)*datasize.KB > conf.Build.MemoryLimit { log.Warningf("[Q] %s->%s exeeds memory limit: %s->%s", pkg.FullRepo, pkg.Pkgbase, datasize.ByteSize(*pkg.DBPackage.MaxRss)*datasize.KB, conf.Build.MemoryLimit) doneQLock.Lock() doneQ = append(doneQ, pkg) doneQLock.Unlock() continue } b.buildingLock.RLock() currentMemLoad := pkgList2MaxMem(b.building) b.buildingLock.RUnlock() // check if package can be build right now if !unknownBuilds && currentMemLoad+(datasize.ByteSize(*pkg.DBPackage.MaxRss)*datasize.KB) > conf.Build.MemoryLimit { log.Debugf("[Q] skipped package with max_rss %s while load %s: %s->%s", datasize.ByteSize(*pkg.DBPackage.MaxRss)*datasize.KB, currentMemLoad, pkg.Pkgbase, pkg.March) continue } } else { b.buildingLock.RLock() if len(b.building) >= MaxUnknownBuilder { b.buildingLock.RUnlock() continue } b.buildingLock.RUnlock() } b.buildingLock.Lock() b.building = append(b.building, pkg) b.buildingLock.Unlock() queueNoMatch = false go func(pkg *ProtoPackage) { dur, err := pkg.build(ctx) if err != nil && !errors.Is(err, ErrorNotEligible) { log.Warningf("[Q] error building package %s->%s in %s: %s", pkg.FullRepo, pkg.Pkgbase, dur, err) b.repoPurge[pkg.FullRepo] <- []*ProtoPackage{pkg} } else if err == nil { log.Infof("[Q] build successful: %s->%s (%s)", pkg.FullRepo, pkg.Pkgbase, dur) } doneQLock.Lock() b.buildingLock.Lock() doneQ = append(doneQ, pkg) for i := 0; i < len(b.building); i++ { if b.building[i].PkgbaseEquals(pkg, true) { b.building = append(b.building[:i], b.building[i+1:]...) break } } doneQLock.Unlock() b.buildingLock.Unlock() b.queueSignal <- struct{}{} }(pkg) } } else { log.Debugf("[Q] memory/build limit reached, waiting for package to finish...") b.buildingLock.RUnlock() <-b.queueSignal queueNoMatch = false } // if only unknown packages are left, enable unknown buildmode b.buildingLock.RLock() if up == len(queue)-(len(doneQ)+len(b.building)) { unknownBuilds = true } b.buildingLock.RUnlock() } return nil } func (b *BuildManager) repoWorker(repo string) { for { select { case pkgL := <-b.repoAdd[repo]: b.repoWG.Add(1) toAdd := make([]string, 0) for _, pkg := range pkgL { toAdd = append(toAdd, pkg.PkgFiles...) } args := []string{"-s", "-v", "-p", "-n", filepath.Join(conf.Basedir.Repo, repo, "os", conf.Arch, repo) + ".db.tar.xz"} args = append(args, toAdd...) cmd := exec.Command("repo-add", args...) res, err := cmd.CombinedOutput() log.Debug(string(res)) if err != nil && cmd.ProcessState.ExitCode() != 1 { log.Panicf("%s while repo-add: %v", string(res), err) } for _, pkg := range pkgL { err = pkg.toDBPackage(true) if err != nil { log.Warningf("error getting db entry for %s: %v", pkg.Pkgbase, err) continue } pkgUpd := pkg.DBPackage.Update(). SetStatus(dbpackage.StatusLatest). ClearSkipReason(). SetRepoVersion(pkg.Version). SetTagRev(pkg.State.TagRev) if _, err := os.Stat(filepath.Join(conf.Basedir.Debug, pkg.March, pkg.DBPackage.Packages[0]+"-debug-"+pkg.Version+"-"+conf.Arch+".pkg.tar.zst")); err == nil { pkgUpd = pkgUpd.SetDebugSymbols(dbpackage.DebugSymbolsAvailable) } else { pkgUpd = pkgUpd.SetDebugSymbols(dbpackage.DebugSymbolsNotAvailable) } pkg.DBPackage = pkgUpd.SaveX(context.Background()) } cmd = exec.Command("paccache", "-rc", filepath.Join(conf.Basedir.Repo, repo, "os", conf.Arch), "-k", "1") //nolint:gosec res, err = cmd.CombinedOutput() log.Debug(string(res)) if err != nil { log.Warningf("error running paccache: %v", err) } err = updateLastUpdated() if err != nil { log.Warningf("error updating lastupdate: %v", err) } b.repoWG.Done() case pkgL := <-b.repoPurge[repo]: for _, pkg := range pkgL { if _, err := os.Stat(filepath.Join(conf.Basedir.Repo, pkg.FullRepo, "os", conf.Arch, pkg.FullRepo) + ".db.tar.xz"); err != nil { continue } if len(pkg.PkgFiles) == 0 { if err := pkg.findPkgFiles(); err != nil { log.Warningf("[%s/%s] unable to find files: %v", pkg.FullRepo, pkg.Pkgbase, err) continue } else if len(pkg.PkgFiles) == 0 { continue } } var realPkgs []string for _, filePath := range pkg.PkgFiles { if _, err := os.Stat(filePath); err == nil { realPkgs = append(realPkgs, Package(filePath).Name()) } } if len(realPkgs) == 0 { continue } b.repoWG.Add(1) args := []string{"-s", "-v", filepath.Join(conf.Basedir.Repo, pkg.FullRepo, "os", conf.Arch, pkg.FullRepo) + ".db.tar.xz"} args = append(args, realPkgs...) cmd := exec.Command("repo-remove", args...) res, err := cmd.CombinedOutput() log.Debug(string(res)) if err != nil && cmd.ProcessState.ExitCode() == 1 { log.Warningf("error while deleting package %s: %s", pkg.Pkgbase, string(res)) } if pkg.DBPackage != nil { _ = pkg.DBPackage.Update().ClearRepoVersion().ClearTagRev().Exec(context.Background()) } for _, file := range pkg.PkgFiles { _ = os.Remove(file) _ = os.Remove(file + ".sig") } err = updateLastUpdated() if err != nil { log.Warningf("error updating lastupdate: %v", err) } b.repoWG.Done() } } } } func (b *BuildManager) syncWorker(ctx context.Context) error { err := os.MkdirAll(conf.Basedir.Work, 0o755) if err != nil { log.Fatalf("error creating work dir %s: %v", conf.Basedir.Work, err) } gitPath := filepath.Join(conf.Basedir.Work, stateDir) for { if _, err := os.Stat(gitPath); os.IsNotExist(err) { cmd := exec.Command("git", "clone", "--depth=1", conf.StateRepo, gitPath) //nolint:gosec res, err := cmd.CombinedOutput() log.Debug(string(res)) if err != nil { log.Fatalf("error cloning state repo: %v", err) } } else if err == nil { cmd := exec.Command("git", "reset", "--hard") cmd.Dir = gitPath res, err := cmd.CombinedOutput() log.Debug(string(res)) if err != nil { log.Fatalf("error reseting state repo: %v", err) } cmd = exec.Command("git", "pull") cmd.Dir = gitPath res, err = cmd.CombinedOutput() log.Debug(string(res)) if err != nil { log.Warningf("failed to update state repo: %v", err) } } // housekeeping wg := new(sync.WaitGroup) for _, repo := range repos { wg.Add(1) splitRepo := strings.Split(repo, "-") repo := repo go func() { err := housekeeping(splitRepo[0], strings.Join(splitRepo[1:], "-"), wg) if err != nil { log.Warningf("[%s] housekeeping failed: %v", repo, err) } }() } wg.Wait() err := logHK() if err != nil { log.Warningf("log-housekeeping failed: %v", err) } debugHK() // fetch updates between sync runs b.alpmMutex.Lock() err = alpmHandle.Release() if err != nil { log.Fatalf("error releasing ALPM handle: %v", err) } if err := retry.Fibonacci(ctx, 1*time.Second, func(_ context.Context) error { if err := setupChroot(); err != nil { log.Warningf("unable to upgrade chroot, trying again later") return retry.RetryableError(err) } return nil }); err != nil { log.Fatal(err) } alpmHandle, err = initALPM(filepath.Join(conf.Basedir.Work, chrootDir, pristineChroot), filepath.Join(conf.Basedir.Work, chrootDir, pristineChroot, "/var/lib/pacman")) if err != nil { log.Warningf("error while alpm-init: %v", err) } b.alpmMutex.Unlock() queue, err := b.genQueue() if err != nil { log.Errorf("error building queue: %v", err) } else { log.Debugf("build-queue with %d items", len(queue)) err = b.buildQueue(ctx, queue) if err != nil { return err } } if ctx.Err() == nil { for _, repo := range repos { err = movePackagesLive(repo) if err != nil { log.Errorf("[%s] error moving packages live: %v", repo, err) } } } else { return ctx.Err() } log.Debugf("build-cycle finished") time.Sleep(time.Duration(*checkInterval) * time.Minute) } } func (b *BuildManager) genQueue() ([]*ProtoPackage, error) { stateFiles, err := Glob(filepath.Join(conf.Basedir.Work, stateDir, "**/*")) if err != nil { return nil, fmt.Errorf("error scanning for state-files: %w", err) } var pkgbuilds []*ProtoPackage for _, stateFile := range stateFiles { stat, err := os.Stat(stateFile) if err != nil || stat.IsDir() || strings.Contains(stateFile, ".git") { continue } repo, subRepo, arch, err := stateFileMeta(stateFile) if err != nil { log.Warningf("[QG] error generating statefile metadata %s: %v", stateFile, err) continue } if !Contains(conf.Repos, repo) || (subRepo != nil && Contains(conf.Blacklist.Repo, *subRepo)) { continue } rawState, err := os.ReadFile(stateFile) if err != nil { log.Warningf("[QG] cannot read statefile %s: %v", stateFile, err) continue } state, err := parseState(string(rawState)) if err != nil { log.Warningf("[QG] cannot parse statefile %s: %v", stateFile, err) continue } for _, march := range conf.March { pkg := &ProtoPackage{ Pkgbase: state.Pkgbase, Repo: dbpackage.Repository(repo), March: march, FullRepo: repo + "-" + march, State: state, Version: state.PkgVer, Arch: arch, } err = pkg.toDBPackage(false) if err != nil { log.Warningf("[QG] error getting/creating dbpackage %s: %v", state.Pkgbase, err) continue } if !pkg.isAvailable(alpmHandle) { log.Debugf("[QG] %s->%s not available on mirror, skipping build", pkg.FullRepo, pkg.Pkgbase) continue } aBuild, err := pkg.IsBuild() if err != nil { log.Warningf("[QG] %s->%s error determining build packages: %v", pkg.FullRepo, pkg.Pkgbase, err) } if aBuild { log.Infof("[QG] %s->%s already build, skipping build", pkg.FullRepo, pkg.Pkgbase) continue } if pkg.DBPackage == nil { err = pkg.toDBPackage(true) if err != nil { log.Warningf("[QG] error getting/creating dbpackage %s: %v", state.Pkgbase, err) continue } } if pkg.DBPackage.TagRev != nil && *pkg.DBPackage.TagRev == state.TagRev { continue } if !pkg.isEligible(context.Background()) { continue } pkg.DBPackage = pkg.DBPackage.Update().SetStatus(dbpackage.StatusQueued).SaveX(context.Background()) pkgbuilds = append(pkgbuilds, pkg) } } return pkgbuilds, nil }