diff --git a/e2e-docker/virtual_test.go b/e2e-docker/virtual_test.go index 42796cc..bc3309b 100644 --- a/e2e-docker/virtual_test.go +++ b/e2e-docker/virtual_test.go @@ -3,9 +3,17 @@ package e2edocker import ( + "bytes" + "compress/gzip" + "encoding/xml" + "fmt" + "io" "net/http" + "os" + "os/exec" "strings" "testing" + "time" ) // TestVirtualPyPIMerge uploads different packages to two pypi locals and @@ -52,3 +60,79 @@ func TestVirtualHelmMerge(t *testing.T) { t.Fatalf("merged helm index missing a member chart (want alpha and beta): %s", s) } } + +// TestVirtualRPMMerge merges a remote rpm repo and a local rpm repo carrying the +// same package: the first member wins the duplicate, its package download +// routes through the virtual, and a real dnf can consume the merged repo. +func TestVirtualRPMMerge(t *testing.T) { + createRepo(t, `{"name":"vrpm-remote","package_type":"rpm","repo_type":"remote","base_url":"`+mockUpstream()+`/rpm-mirror","stale_on_error":true}`) + createRepo(t, `{"name":"vrpm-local","package_type":"rpm","repo_type":"local"}`) + defer deleteRepo(t, "vrpm-remote") + defer deleteRepo(t, "vrpm-local") + + pkg := fixtureBytes(t, "rpmrepo/Packages/e2e-testpkg-1.0-1.noarch.rpm") + uploadFile(t, "vrpm-local", "e2e-testpkg-1.0-1.noarch.rpm", pkg, "application/x-rpm") + + createVirtual(t, `{"name":"vrpm","package_type":"rpm","members":["vrpm-remote","vrpm-local"]}`) + defer deleteVirtual(t, "vrpm") + + resp, body := getEventually(t, api("/api/v1/virtual/vrpm/repodata/repomd.xml"), 15*time.Second) + if resp.StatusCode != http.StatusOK { + t.Fatalf("virtual repomd.xml: status %d: %s", resp.StatusCode, body) + } + var md struct { + Data []struct { + Type string `xml:"type,attr"` + Location struct { + Href string `xml:"href,attr"` + } `xml:"location"` + } `xml:"data"` + } + if err := xml.Unmarshal(body, &md); err != nil { + t.Fatalf("parse repomd.xml: %v\n%s", err, body) + } + var primary []byte + for _, d := range md.Data { + if d.Type == "primary" { + resp, gz := doRequest(t, http.MethodGet, api("/api/v1/virtual/vrpm/"+d.Location.Href), nil, "") + if resp.StatusCode != http.StatusOK { + t.Fatalf("primary: status %d", resp.StatusCode) + } + r, err := gzip.NewReader(bytes.NewReader(gz)) + if err != nil { + t.Fatal(err) + } + primary, _ = io.ReadAll(r) + } + } + href := "vrpm-remote/Packages/e2e-testpkg-1.0-1.noarch.rpm" + if strings.Count(string(primary), "e2e-testpkg") != 1 || !strings.Contains(string(primary), `href="`+href+`"`) { + t.Fatalf("merged primary should hold one e2e-testpkg owned by the first member:\n%s", primary) + } + + resp, got := doRequest(t, http.MethodGet, api("/api/v1/virtual/vrpm/"+href), nil, "") + if resp.StatusCode != http.StatusOK || !bytes.Equal(got, pkg) { + t.Fatalf("package via virtual: status %d, %d bytes (want %d)", resp.StatusCode, len(got), len(pkg)) + } + + network := os.Getenv("COMPOSE_NETWORK") + internal := os.Getenv("ARTIFACTAPI_INTERNAL") + if network == "" || internal == "" { + t.Skip("COMPOSE_NETWORK/ARTIFACTAPI_INTERNAL not set; skipping real dnf") + } + if _, err := exec.LookPath("docker"); err != nil { + t.Skip("docker not available on the test host") + } + repoConf := fmt.Sprintf("[vrpm]\nname=vrpm\nbaseurl=%s/api/v1/virtual/vrpm/\nenabled=1\ngpgcheck=0\nmetadata_expire=0\n", strings.TrimRight(internal, "/")) + script := "set -euo pipefail; " + + "printf '%s' \"$REPO\" > /etc/yum.repos.d/vrpm.repo; " + + "dnf -y --disablerepo='*' --enablerepo=vrpm makecache; " + + "dnf -y --disablerepo='*' --enablerepo=vrpm list --available; " + + "dnf -y --disablerepo='*' --enablerepo=vrpm install e2e-testpkg; " + + "rpm -q e2e-testpkg" + out, err := exec.Command("docker", "run", "--rm", "--network", network, "-e", "REPO="+repoConf, + "rockylinux:9", "bash", "-c", script).CombinedOutput() + if err != nil { + t.Fatalf("real dnf against rpm virtual failed: %v\n%s", err, out) + } +} diff --git a/internal/api/v1/proxy.go b/internal/api/v1/proxy.go index bb0474b..a056057 100644 --- a/internal/api/v1/proxy.go +++ b/internal/api/v1/proxy.go @@ -162,7 +162,16 @@ func (h *ProxyHandler) handleVirtual(w http.ResponseWriter, r *http.Request) { proxyBaseURL := fmt.Sprintf("%s://%s", scheme(r), r.Host) + if loc, ok := h.virtualEngine.MemberRedirect(r.Context(), *virt, path, proxyBaseURL); ok { + http.Redirect(w, r, loc, http.StatusFound) + return + } + body, contentType, err := h.virtualEngine.Fetch(r.Context(), *virt, path, proxyBaseURL) + if errors.Is(err, virtual.ErrNotFound) { + http.Error(w, "not found", http.StatusNotFound) + return + } if err != nil { slog.Error("virtual fetch failed", "virtual", virtualName, "path", path, "error", err) http.Error(w, "bad gateway", http.StatusBadGateway) diff --git a/internal/virtual/engine.go b/internal/virtual/engine.go index bcc352e..0539493 100644 --- a/internal/virtual/engine.go +++ b/internal/virtual/engine.go @@ -2,9 +2,14 @@ package virtual import ( "context" + "errors" "fmt" "io" "log/slog" + "net/http" + "net/http/httptest" + "slices" + "strings" "sync" "git.unkin.net/unkin/artifactapi/internal/database" @@ -23,6 +28,10 @@ func NewEngine(db *database.DB, proxyEngine *proxy.Engine) *Engine { } func (e *Engine) Fetch(ctx context.Context, virt models.Virtual, path string, proxyBaseURL string) ([]byte, string, error) { + if virt.PackageType == models.PackageRPM { + return e.fetchRPM(ctx, virt, path) + } + merger, err := GetMerger(virt.PackageType) if err != nil { return nil, "", fmt.Errorf("unsupported virtual type %q: %w", virt.PackageType, err) @@ -133,3 +142,137 @@ func (e *Engine) fetchLocalIndex(ctx context.Context, remote models.Remote, path return indexer.GenerateLocalIndex(ctx, e.db, remote.Name, path) } + +var ErrNotFound = errors.New("not found") + +// MemberRedirect maps an rpm virtual package path (/, as written +// by MergeRPM) to the owning member's route. ok is false for any other path. +func (e *Engine) MemberRedirect(ctx context.Context, virt models.Virtual, path, proxyBaseURL string) (string, bool) { + if virt.PackageType != models.PackageRPM || strings.HasPrefix(path, "repodata/") { + return "", false + } + name, rest, found := strings.Cut(path, "/") + if !found || !slices.Contains(virt.Members, name) { + return "", false + } + remote, err := e.db.GetRemote(ctx, name) + if err != nil { + return "", false + } + return fmt.Sprintf("%s/api/v1/%s/%s/%s", strings.TrimRight(proxyBaseURL, "/"), remote.RepoType, name, rest), true +} + +func (e *Engine) fetchRPM(ctx context.Context, virt models.Virtual, path string) ([]byte, string, error) { + if !strings.HasPrefix(path, "repodata/") { + return nil, "", ErrNotFound + } + + results := make([]*RPMMember, len(virt.Members)) + var wg sync.WaitGroup + for i, name := range virt.Members { + wg.Add(1) + go func() { + defer wg.Done() + m, err := e.fetchRPMMember(ctx, name) + if err != nil { + slog.Warn("virtual member fetch failed", "virtual", virt.Name, "member", name, "error", err) + return + } + results[i] = m + }() + } + wg.Wait() + + var members []RPMMember + for _, m := range results { + if m != nil { + members = append(members, *m) + } + } + if len(members) == 0 { + return nil, "", fmt.Errorf("no members reachable for virtual %q", virt.Name) + } + + // ponytail: merges on every repodata request; cache keyed on member repomd checksums if member count/size makes this slow. + repo, err := MergeRPM(members) + if err != nil { + return nil, "", fmt.Errorf("merge rpm repodata: %w", err) + } + if path == "repodata/repomd.xml" { + return repo.Repomd, "application/xml", nil + } + if body, ok := repo.Files[path]; ok { + return body, "application/gzip", nil + } + return nil, "", ErrNotFound +} + +func (e *Engine) fetchRPMMember(ctx context.Context, name string) (*RPMMember, error) { + remote, err := e.db.GetRemote(ctx, name) + if err != nil { + return nil, fmt.Errorf("remote %q: %w", name, err) + } + repomd, err := e.fetchMemberPath(ctx, *remote, "repodata/repomd.xml") + if err != nil { + return nil, err + } + locs, ts, err := parseRepomd(repomd) + if err != nil { + return nil, err + } + m := &RPMMember{RemoteName: name, Timestamp: ts, Data: map[string][]byte{}} + for t, href := range locs { + raw, err := e.fetchMemberPath(ctx, *remote, href) + if err != nil { + return nil, err + } + if m.Data[t], err = decompress(href, raw); err != nil { + return nil, fmt.Errorf("decompress %s: %w", href, err) + } + } + return m, nil +} + +// fetchMemberPath reads one path from a member the same way its own route +// would serve it: local index, synthesized remote (e.g. github_rpm), or proxy. +func (e *Engine) fetchMemberPath(ctx context.Context, remote models.Remote, path string) ([]byte, error) { + prov, err := provider.Get(remote.PackageType) + if err != nil { + return nil, fmt.Errorf("provider %q: %w", remote.PackageType, err) + } + + var serve func(w http.ResponseWriter, r *http.Request) bool + if remote.RepoType == models.RepoTypeLocal { + indexer, ok := prov.(provider.LocalIndexer) + if !ok { + return nil, fmt.Errorf("provider %q does not serve a local index", remote.PackageType) + } + serve = func(w http.ResponseWriter, r *http.Request) bool { + return indexer.ServeLocalIndex(w, r, e.db, remote.Name, path) + } + } else if rs, ok := prov.(provider.RemoteServer); ok { + serve = func(w http.ResponseWriter, r *http.Request) bool { + return rs.ServeRemote(w, r, remote, path, "", e.db) + } + } + + if serve != nil { + rec := httptest.NewRecorder() + if serve(rec, httptest.NewRequestWithContext(ctx, http.MethodGet, "/"+path, nil)) { + if rec.Code != http.StatusOK { + return nil, fmt.Errorf("%s/%s: status %d", remote.Name, path, rec.Code) + } + return rec.Body.Bytes(), nil + } + if remote.RepoType == models.RepoTypeLocal { + return nil, fmt.Errorf("%s/%s: %w", remote.Name, path, ErrNotFound) + } + } + + res, err := e.proxyEngine.Fetch(ctx, remote, path, prov) + if err != nil { + return nil, fmt.Errorf("fetch %s/%s: %w", remote.Name, path, err) + } + defer res.Reader.Close() + return io.ReadAll(res.Reader) +}