ALHP.GO/buildmanager.go

458 lines
13 KiB
Go

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 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 {
if pkg.DBPackage != nil {
_ = pkg.DBPackage.Update().ClearRepoVersion().ClearTagRev().Exec(context.Background())
}
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, "-")
go func() { //nolint:contextcheck
err := housekeeping(splitRepo[0], strings.Join(splitRepo[1:], "-"), wg)
if err != nil {
log.Warningf("[%s] housekeeping failed: %v", repo, err)
}
}()
}
wg.Wait()
err := logHK() //nolint:contextcheck
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() //nolint:contextcheck
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) //nolint:contextcheck
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") || strings.Contains(stateFile, "README.md") {
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.IsBuilt()
if err != nil {
log.Warningf("[QG] %s->%s error determining built packages: %v", pkg.FullRepo, pkg.Pkgbase, err)
}
if aBuild {
log.Infof("[QG] %s->%s already built, 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
}
// try download srcinfo from repo
srcInfo, err := downloadSRCINFO(pkg.DBPackage.Pkgbase, state.TagRev)
if err == nil {
pkg.Srcinfo = srcInfo
}
if !pkg.isEligible(context.Background()) {
continue
}
pkg.DBPackage = pkg.DBPackage.Update().SetStatus(dbpackage.StatusQueued).SaveX(context.Background())
pkgbuilds = append(pkgbuilds, pkg)
}
}
return pkgbuilds, nil
}