remotes: support multiple base_urls with round-robin + failover
A remote's base_url may now be a single string OR a list of upstream mirrors. When it is a list the shared proxy engine load-balances across them round-robin and, on an upstream error/timeout/5xx, fails over to the next mirror before returning an error. Because the selection happens in the engine (not per provider), it applies to every remote package type. Backward compatible: a bare-string base_url behaves exactly as before. - add models.StringOrSlice (string-or-array JSON) and custom Remote (Un)MarshalJSON: base_url populates BaseURLs (full list) + BaseURL (active/first); marshals a single mirror back to a bare string - add Remote.BaseURLList / ValidateBaseURLs; validate list is non-empty and every entry is an http/https URL in the v2 create/update handlers - persist the full list in a new base_urls TEXT[] column (additive migration), keeping base_url in sync for old readers; only write base_urls for genuinely multi-mirror remotes - engine: per-remote round-robin cursor + attempt ordering; wrap the fetch/head/revalidate upstream calls in a failover loop that narrows the remote to one selected mirror per attempt; only network errors and 5xx fail over (404/403/... are returned as-is); circuit breaker stays keyed per remote and trips only after all mirrors fail - add unit tests (JSON round-trip, engine round-robin/failover/single-URL, DB multi-URL round-trip) and a docker acceptance suite: round-robin distribution across two mock upstreams, failover past a dead primary, single-base_url regression, and a real dnf makecache+install through a two-mirror rpm remote whose primary is dead Least-connections and a per-remote strategy selector are a follow-up PR.
This commit is contained in:
@@ -74,8 +74,8 @@ func (h *RemotesHandler) create(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, fmt.Sprintf("invalid repo type: %q", remote.RepoType), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if remote.RepoType == models.RepoTypeRemote && remote.BaseURL == "" {
|
||||
http.Error(w, "base_url is required for remote repositories", http.StatusBadRequest)
|
||||
if err := remote.ValidateBaseURLs(); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if err := remote.ValidatePatterns(); err != nil {
|
||||
@@ -102,6 +102,10 @@ func (h *RemotesHandler) update(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
remote.Name = name
|
||||
if err := remote.ValidateBaseURLs(); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if err := remote.ValidatePatterns(); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
|
||||
@@ -99,6 +99,40 @@ func TestRemotesCRUD(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRemoteMultiBaseURLRoundTrip(t *testing.T) {
|
||||
requireDB(t)
|
||||
urls := []string{"https://a.example", "https://b.example"}
|
||||
if err := testDB.CreateRemote(ctx(), &models.Remote{
|
||||
Name: "r-multi", PackageType: models.PackageRPM, RepoType: models.RepoTypeRemote,
|
||||
BaseURLs: urls, MutableTTL: 3600,
|
||||
}); err != nil {
|
||||
t.Fatalf("create multi-url remote: %v", err)
|
||||
}
|
||||
defer testDB.DeleteRemote(ctx(), "r-multi")
|
||||
|
||||
got, err := testDB.GetRemote(ctx(), "r-multi")
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if len(got.BaseURLs) != 2 || got.BaseURLs[0] != urls[0] || got.BaseURLs[1] != urls[1] {
|
||||
t.Fatalf("BaseURLs round-trip = %v, want %v", got.BaseURLs, urls)
|
||||
}
|
||||
if got.BaseURL != urls[0] {
|
||||
t.Fatalf("active BaseURL = %q, want %q", got.BaseURL, urls[0])
|
||||
}
|
||||
|
||||
// Reducing to a single URL clears the multi-URL column on update.
|
||||
got.BaseURLs = []string{"https://only.example"}
|
||||
got.BaseURL = "https://only.example"
|
||||
if err := testDB.UpdateRemote(ctx(), got); err != nil {
|
||||
t.Fatalf("update to single: %v", err)
|
||||
}
|
||||
got, _ = testDB.GetRemote(ctx(), "r-multi")
|
||||
if len(got.BaseURLs) != 0 || got.BaseURL != "https://only.example" {
|
||||
t.Fatalf("single-URL update = %v / %q, want empty list and single base_url", got.BaseURLs, got.BaseURL)
|
||||
}
|
||||
}
|
||||
|
||||
func TestArtifactsAndBlobs(t *testing.T) {
|
||||
requireDB(t)
|
||||
seedRemote(t, "r-art")
|
||||
|
||||
@@ -44,6 +44,7 @@ func (db *DB) migrate() error {
|
||||
package_type TEXT NOT NULL,
|
||||
repo_type TEXT DEFAULT 'remote',
|
||||
base_url TEXT NOT NULL DEFAULT '',
|
||||
base_urls TEXT[] DEFAULT '{}',
|
||||
description TEXT DEFAULT '',
|
||||
username TEXT DEFAULT '',
|
||||
password TEXT DEFAULT '',
|
||||
@@ -124,6 +125,7 @@ func (db *DB) migrate() error {
|
||||
CREATE INDEX IF NOT EXISTS idx_access_log_remote_time ON access_log(remote_name, created_at);
|
||||
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS repo_type TEXT DEFAULT 'remote';
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS base_urls TEXT[] DEFAULT '{}';
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS upstream_dial_timeout INTEGER DEFAULT 0;
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS upstream_tls_timeout INTEGER DEFAULT 0;
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS upstream_response_header_timeout INTEGER DEFAULT 0;
|
||||
|
||||
@@ -6,7 +6,7 @@ import (
|
||||
"git.unkin.net/unkin/artifactapi/pkg/models"
|
||||
)
|
||||
|
||||
const remoteCols = `name, package_type, repo_type, base_url, description, username, password,
|
||||
const remoteCols = `name, package_type, repo_type, base_url, base_urls, description, username, password,
|
||||
immutable_ttl, mutable_ttl, check_mutable,
|
||||
patterns, blocklist, mutable_patterns, immutable_patterns,
|
||||
ban_tags_enabled, ban_tags,
|
||||
@@ -16,8 +16,10 @@ const remoteCols = `name, package_type, repo_type, base_url, description, userna
|
||||
created_at, updated_at`
|
||||
|
||||
func scanRemote(scanner interface{ Scan(...any) error }, r *models.Remote) error {
|
||||
return scanner.Scan(
|
||||
&r.Name, &r.PackageType, &r.RepoType, &r.BaseURL, &r.Description, &r.Username, &r.Password,
|
||||
var baseURL string
|
||||
var baseURLs []string
|
||||
if err := scanner.Scan(
|
||||
&r.Name, &r.PackageType, &r.RepoType, &baseURL, &baseURLs, &r.Description, &r.Username, &r.Password,
|
||||
&r.ImmutableTTL, &r.MutableTTL, &r.CheckMutable,
|
||||
&r.Patterns, &r.Blocklist, &r.MutablePatterns, &r.ImmutablePatterns,
|
||||
&r.BanTagsEnabled, &r.BanTags,
|
||||
@@ -25,7 +27,19 @@ func scanRemote(scanner interface{ Scan(...any) error }, r *models.Remote) error
|
||||
&r.ReleasesRemote, &r.ManagedBy,
|
||||
&r.UpstreamDialTimeout, &r.UpstreamTLSTimeout, &r.UpstreamResponseHeaderTimeout,
|
||||
&r.CreatedAt, &r.UpdatedAt,
|
||||
)
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
// Prefer the full list; fall back to the legacy single column so remotes
|
||||
// written before base_urls existed still resolve an upstream.
|
||||
if len(baseURLs) > 0 {
|
||||
r.BaseURLs = baseURLs
|
||||
r.BaseURL = baseURLs[0]
|
||||
} else {
|
||||
r.BaseURLs = nil
|
||||
r.BaseURL = baseURL
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (db *DB) GetRemote(ctx context.Context, name string) (*models.Remote, error) {
|
||||
@@ -58,16 +72,16 @@ func (db *DB) ListRemotes(ctx context.Context) ([]models.Remote, error) {
|
||||
func (db *DB) CreateRemote(ctx context.Context, r *models.Remote) error {
|
||||
_, err := db.Pool.Exec(ctx, `
|
||||
INSERT INTO remotes (
|
||||
name, package_type, repo_type, base_url, description, username, password,
|
||||
name, package_type, repo_type, base_url, base_urls, description, username, password,
|
||||
immutable_ttl, mutable_ttl, check_mutable,
|
||||
patterns, blocklist, mutable_patterns, immutable_patterns,
|
||||
ban_tags_enabled, ban_tags,
|
||||
quarantine_enabled, quarantine_days, stale_on_error,
|
||||
releases_remote, managed_by,
|
||||
upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout
|
||||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24)
|
||||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25)
|
||||
`,
|
||||
r.Name, r.PackageType, r.RepoType, r.BaseURL, r.Description, r.Username, r.Password,
|
||||
r.Name, r.PackageType, r.RepoType, firstBaseURL(r), baseURLsColumn(r), r.Description, r.Username, r.Password,
|
||||
r.ImmutableTTL, r.MutableTTL, r.CheckMutable,
|
||||
r.Patterns, r.Blocklist, r.MutablePatterns, r.ImmutablePatterns,
|
||||
r.BanTagsEnabled, r.BanTags,
|
||||
@@ -78,10 +92,30 @@ func (db *DB) CreateRemote(ctx context.Context, r *models.Remote) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// firstBaseURL returns the active/first upstream URL for the legacy base_url
|
||||
// column, kept in sync with base_urls so pre-multi-URL readers still work.
|
||||
func firstBaseURL(r *models.Remote) string {
|
||||
list := r.BaseURLList()
|
||||
if len(list) == 0 {
|
||||
return ""
|
||||
}
|
||||
return list[0]
|
||||
}
|
||||
|
||||
// baseURLsColumn returns the value for the base_urls column. It is only written
|
||||
// for genuinely multi-mirror remotes; single-URL remotes leave it empty and rely
|
||||
// on base_url, so the common "read, mutate BaseURL, update" flow keeps working.
|
||||
func baseURLsColumn(r *models.Remote) []string {
|
||||
if list := r.BaseURLList(); len(list) > 1 {
|
||||
return list
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (db *DB) UpdateRemote(ctx context.Context, r *models.Remote) error {
|
||||
_, err := db.Pool.Exec(ctx, `
|
||||
UPDATE remotes SET
|
||||
package_type=$2, repo_type=$3, base_url=$4, description=$5, username=$6, password=$7,
|
||||
package_type=$2, repo_type=$3, base_url=$4, base_urls=$25, description=$5, username=$6, password=$7,
|
||||
immutable_ttl=$8, mutable_ttl=$9, check_mutable=$10,
|
||||
patterns=$11, blocklist=$12, mutable_patterns=$13, immutable_patterns=$14,
|
||||
ban_tags_enabled=$15, ban_tags=$16,
|
||||
@@ -91,13 +125,14 @@ func (db *DB) UpdateRemote(ctx context.Context, r *models.Remote) error {
|
||||
updated_at=NOW()
|
||||
WHERE name=$1
|
||||
`,
|
||||
r.Name, r.PackageType, r.RepoType, r.BaseURL, r.Description, r.Username, r.Password,
|
||||
r.Name, r.PackageType, r.RepoType, firstBaseURL(r), r.Description, r.Username, r.Password,
|
||||
r.ImmutableTTL, r.MutableTTL, r.CheckMutable,
|
||||
r.Patterns, r.Blocklist, r.MutablePatterns, r.ImmutablePatterns,
|
||||
r.BanTagsEnabled, r.BanTags,
|
||||
r.QuarantineEnabled, r.QuarantineDays, r.StaleOnError,
|
||||
r.ReleasesRemote, r.ManagedBy,
|
||||
r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout,
|
||||
baseURLsColumn(r),
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -11,6 +11,8 @@ import (
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/internal/cache"
|
||||
@@ -35,6 +37,10 @@ type Engine struct {
|
||||
cas *storage.CAS
|
||||
circuit *CircuitBreaker
|
||||
accessLog chan database.AccessLogEntry
|
||||
// rrCounters holds a per-remote round-robin cursor (remoteName ->
|
||||
// *atomic.Uint64) used to rotate the starting mirror across upstream base
|
||||
// URLs. Distribution is per-replica and approximate, which is fine.
|
||||
rrCounters sync.Map
|
||||
}
|
||||
|
||||
func NewEngine(db *database.DB, c *cache.Redis, s *storage.S3) *Engine {
|
||||
@@ -222,7 +228,30 @@ func (e *Engine) Head(ctx context.Context, remote models.Remote, path string, pr
|
||||
return e.headUpstream(ctx, remote, path, prov)
|
||||
}
|
||||
|
||||
// headUpstream issues an upstream HEAD, load-balancing across the remote's base
|
||||
// URLs and failing over to the next mirror on a network error or 5xx.
|
||||
func (e *Engine) headUpstream(ctx context.Context, remote models.Remote, path string, prov provider.Provider) (*HeadResult, error) {
|
||||
order := e.baseURLAttemptOrder(remote)
|
||||
if len(order) == 0 {
|
||||
return nil, &ProxyError{Status: http.StatusBadGateway, Message: "no upstream base_url configured"}
|
||||
}
|
||||
var lastErr error
|
||||
for i, url := range order {
|
||||
result, err := e.headUpstreamOnce(ctx, withBaseURL(remote, url), path, prov)
|
||||
if err == nil {
|
||||
return result, nil
|
||||
}
|
||||
lastErr = err
|
||||
if i < len(order)-1 && shouldFailover(err) {
|
||||
slog.Warn("upstream HEAD failed, failing over", "remote", remote.Name, "base_url", url, "error", err)
|
||||
continue
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return nil, lastErr
|
||||
}
|
||||
|
||||
func (e *Engine) headUpstreamOnce(ctx context.Context, remote models.Remote, path string, prov provider.Provider) (*HeadResult, error) {
|
||||
url := prov.UpstreamURL(remote, path)
|
||||
|
||||
authHeaders, err := prov.AuthHeaders(ctx, remote)
|
||||
@@ -277,7 +306,31 @@ func (e *Engine) headUpstream(ctx context.Context, remote models.Remote, path st
|
||||
return &HeadResult{ContentType: contentType, Size: resp.ContentLength, Source: "remote"}, nil
|
||||
}
|
||||
|
||||
// fetchFromUpstream fetches an artifact from upstream, load-balancing across the
|
||||
// remote's base URLs and failing over to the next mirror on a network error or
|
||||
// 5xx before returning an error.
|
||||
func (e *Engine) fetchFromUpstream(ctx context.Context, remote models.Remote, path string, prov provider.Provider, class Classification, ttl time.Duration, clientHeaders http.Header) (*FetchResult, error) {
|
||||
order := e.baseURLAttemptOrder(remote)
|
||||
if len(order) == 0 {
|
||||
return nil, &ProxyError{Status: http.StatusBadGateway, Message: "no upstream base_url configured"}
|
||||
}
|
||||
var lastErr error
|
||||
for i, url := range order {
|
||||
result, err := e.fetchFromUpstreamOnce(ctx, withBaseURL(remote, url), path, prov, class, ttl, clientHeaders)
|
||||
if err == nil {
|
||||
return result, nil
|
||||
}
|
||||
lastErr = err
|
||||
if i < len(order)-1 && shouldFailover(err) {
|
||||
slog.Warn("upstream fetch failed, failing over", "remote", remote.Name, "base_url", url, "error", err)
|
||||
continue
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return nil, lastErr
|
||||
}
|
||||
|
||||
func (e *Engine) fetchFromUpstreamOnce(ctx context.Context, remote models.Remote, path string, prov provider.Provider, class Classification, ttl time.Duration, clientHeaders http.Header) (*FetchResult, error) {
|
||||
url := prov.UpstreamURL(remote, path)
|
||||
|
||||
authHeaders, err := prov.AuthHeaders(ctx, remote)
|
||||
@@ -454,7 +507,31 @@ func (e *Engine) serveFromStore(ctx context.Context, remote models.Remote, path
|
||||
}, nil
|
||||
}
|
||||
|
||||
// checkUpstream issues a conditional upstream HEAD (If-None-Match), load
|
||||
// balancing across the remote's base URLs and failing over to the next mirror on
|
||||
// a network error or 5xx.
|
||||
func (e *Engine) checkUpstream(ctx context.Context, remote models.Remote, path, etag string, prov provider.Provider) (bool, error) {
|
||||
order := e.baseURLAttemptOrder(remote)
|
||||
if len(order) == 0 {
|
||||
return false, &ProxyError{Status: http.StatusBadGateway, Message: "no upstream base_url configured"}
|
||||
}
|
||||
var lastErr error
|
||||
for i, url := range order {
|
||||
notModified, err := e.checkUpstreamOnce(ctx, withBaseURL(remote, url), path, etag, prov)
|
||||
if err == nil {
|
||||
return notModified, nil
|
||||
}
|
||||
lastErr = err
|
||||
if i < len(order)-1 && shouldFailover(err) {
|
||||
slog.Warn("upstream revalidation failed, failing over", "remote", remote.Name, "base_url", url, "error", err)
|
||||
continue
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
return false, lastErr
|
||||
}
|
||||
|
||||
func (e *Engine) checkUpstreamOnce(ctx context.Context, remote models.Remote, path, etag string, prov provider.Provider) (bool, error) {
|
||||
url := prov.UpstreamURL(remote, path)
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodHead, url, nil)
|
||||
@@ -649,3 +726,44 @@ func isNetworkError(err error) bool {
|
||||
var ue *UpstreamError
|
||||
return errors.As(err, &ue)
|
||||
}
|
||||
|
||||
// baseURLAttemptOrder returns the ordered upstream base URLs to try for a single
|
||||
// request. A multi-mirror remote starts at the next round-robin position and
|
||||
// advances linearly for failover; a single-URL remote yields exactly that one
|
||||
// URL, preserving the original one-attempt behavior.
|
||||
func (e *Engine) baseURLAttemptOrder(remote models.Remote) []string {
|
||||
urls := remote.BaseURLList()
|
||||
if len(urls) <= 1 {
|
||||
return urls
|
||||
}
|
||||
v, _ := e.rrCounters.LoadOrStore(remote.Name, new(atomic.Uint64))
|
||||
start := int(v.(*atomic.Uint64).Add(1) - 1)
|
||||
ordered := make([]string, len(urls))
|
||||
for i := range urls {
|
||||
ordered[i] = urls[(start+i)%len(urls)]
|
||||
}
|
||||
return ordered
|
||||
}
|
||||
|
||||
// withBaseURL narrows a remote's active BaseURL to a single selected mirror so
|
||||
// providers (UpstreamURL/AuthHeaders/RewriteResponse) operate on exactly that
|
||||
// upstream for this attempt.
|
||||
func withBaseURL(remote models.Remote, url string) models.Remote {
|
||||
remote.BaseURL = url
|
||||
remote.BaseURLs = []string{url}
|
||||
return remote
|
||||
}
|
||||
|
||||
// shouldFailover reports whether an upstream attempt error is worth retrying
|
||||
// against the next mirror: network errors/timeouts and upstream 5xx responses.
|
||||
// Definitive statuses (404/403/401/...) are returned to the caller unchanged.
|
||||
func shouldFailover(err error) bool {
|
||||
if isNetworkError(err) {
|
||||
return true
|
||||
}
|
||||
var pe *ProxyError
|
||||
if errors.As(err, &pe) {
|
||||
return pe.Status >= 500
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -0,0 +1,184 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/pkg/models"
|
||||
)
|
||||
|
||||
// TestFetchMultiBaseURLRoundRobin drives distinct artifact paths through a
|
||||
// remote configured with two upstreams and asserts both receive traffic.
|
||||
func TestFetchMultiBaseURLRoundRobin(t *testing.T) {
|
||||
requireStack(t)
|
||||
ctx := context.Background()
|
||||
|
||||
var hitsA, hitsB atomic.Int64
|
||||
upA := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
hitsA.Add(1)
|
||||
w.Write([]byte("A"))
|
||||
}))
|
||||
defer upA.Close()
|
||||
upB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
hitsB.Add(1)
|
||||
w.Write([]byte("B"))
|
||||
}))
|
||||
defer upB.Close()
|
||||
|
||||
r := seed(t, models.Remote{
|
||||
Name: "eng-rr",
|
||||
PackageType: models.PackageGeneric,
|
||||
RepoType: models.RepoTypeRemote,
|
||||
BaseURLs: []string{upA.URL, upB.URL},
|
||||
StaleOnError: true,
|
||||
})
|
||||
p := prov(t, models.PackageGeneric)
|
||||
|
||||
const n = 10
|
||||
for i := 0; i < n; i++ {
|
||||
res, err := testEngine.Fetch(ctx, r, fmt.Sprintf("rr-%d.bin", i), p)
|
||||
if err != nil {
|
||||
t.Fatalf("fetch %d: %v", i, err)
|
||||
}
|
||||
res.Reader.Close()
|
||||
}
|
||||
|
||||
if hitsA.Load() == 0 || hitsB.Load() == 0 {
|
||||
t.Fatalf("round-robin did not spread across both upstreams: A=%d B=%d", hitsA.Load(), hitsB.Load())
|
||||
}
|
||||
if total := hitsA.Load() + hitsB.Load(); total != n {
|
||||
t.Fatalf("expected %d upstream hits total, got %d (A=%d B=%d)", n, total, hitsA.Load(), hitsB.Load())
|
||||
}
|
||||
}
|
||||
|
||||
// TestFetchMultiBaseURLFailover asserts that a dead/erroring primary mirror
|
||||
// transparently fails over to a healthy secondary, for both a 5xx primary and a
|
||||
// network-unreachable primary.
|
||||
func TestFetchMultiBaseURLFailover(t *testing.T) {
|
||||
requireStack(t)
|
||||
ctx := context.Background()
|
||||
|
||||
var hitsB atomic.Int64
|
||||
upB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
hitsB.Add(1)
|
||||
w.Write([]byte("served-by-B"))
|
||||
}))
|
||||
defer upB.Close()
|
||||
up500 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
}))
|
||||
defer up500.Close()
|
||||
|
||||
p := prov(t, models.PackageGeneric)
|
||||
|
||||
// Primary returns 5xx: every request must still succeed via the secondary.
|
||||
r5xx := seed(t, models.Remote{
|
||||
Name: "eng-failover-5xx",
|
||||
PackageType: models.PackageGeneric,
|
||||
RepoType: models.RepoTypeRemote,
|
||||
BaseURLs: []string{up500.URL, upB.URL},
|
||||
})
|
||||
for i := 0; i < 6; i++ {
|
||||
res, err := testEngine.Fetch(ctx, r5xx, fmt.Sprintf("fo5-%d.bin", i), p)
|
||||
if err != nil {
|
||||
t.Fatalf("5xx failover fetch %d: %v", i, err)
|
||||
}
|
||||
if got := readAll(t, res); got != "served-by-B" {
|
||||
t.Fatalf("5xx failover fetch %d body=%q, want served-by-B", i, got)
|
||||
}
|
||||
}
|
||||
|
||||
// Primary is network-unreachable: failover must still reach the secondary.
|
||||
rNet := seed(t, models.Remote{
|
||||
Name: "eng-failover-net",
|
||||
PackageType: models.PackageGeneric,
|
||||
RepoType: models.RepoTypeRemote,
|
||||
BaseURLs: []string{"http://127.0.0.1:1", upB.URL},
|
||||
})
|
||||
res, err := testEngine.Fetch(ctx, rNet, "fonet.bin", p)
|
||||
if err != nil {
|
||||
t.Fatalf("network failover fetch: %v", err)
|
||||
}
|
||||
if got := readAll(t, res); got != "served-by-B" {
|
||||
t.Fatalf("network failover body=%q, want served-by-B", got)
|
||||
}
|
||||
if hitsB.Load() == 0 {
|
||||
t.Fatal("secondary upstream never served during failover")
|
||||
}
|
||||
}
|
||||
|
||||
// TestFetchDefinitiveStatusNoFailover asserts a definitive 404 from the first
|
||||
// mirror is returned as-is (not failed over): a missing artifact is not a mirror
|
||||
// outage. The remote is fresh so its round-robin cursor starts at index 0.
|
||||
func TestFetchDefinitiveStatusNoFailover(t *testing.T) {
|
||||
requireStack(t)
|
||||
ctx := context.Background()
|
||||
|
||||
var hitsB atomic.Int64
|
||||
up404 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
http.NotFound(w, r)
|
||||
}))
|
||||
defer up404.Close()
|
||||
upB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
hitsB.Add(1)
|
||||
w.Write([]byte("B"))
|
||||
}))
|
||||
defer upB.Close()
|
||||
|
||||
r := seed(t, models.Remote{
|
||||
Name: "eng-no-failover-404",
|
||||
PackageType: models.PackageGeneric,
|
||||
RepoType: models.RepoTypeRemote,
|
||||
BaseURLs: []string{up404.URL, upB.URL},
|
||||
})
|
||||
_, err := testEngine.Fetch(ctx, r, "missing.bin", prov(t, models.PackageGeneric))
|
||||
var pe *ProxyError
|
||||
if err == nil || !asProxyError(err, &pe) || pe.Status != http.StatusNotFound {
|
||||
t.Fatalf("expected 404 ProxyError without failover, got %v", err)
|
||||
}
|
||||
if hitsB.Load() != 0 {
|
||||
t.Fatalf("404 from primary must not fail over, but secondary was hit %d times", hitsB.Load())
|
||||
}
|
||||
}
|
||||
|
||||
// TestFetchSingleBaseURLUnchanged asserts a single-URL remote behaves exactly as
|
||||
// before: one healthy URL succeeds, and one dead URL errors with no failover.
|
||||
func TestFetchSingleBaseURLUnchanged(t *testing.T) {
|
||||
requireStack(t)
|
||||
ctx := context.Background()
|
||||
|
||||
upB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Write([]byte("solo"))
|
||||
}))
|
||||
defer upB.Close()
|
||||
|
||||
p := prov(t, models.PackageGeneric)
|
||||
|
||||
rOK := seed(t, models.Remote{
|
||||
Name: "eng-solo",
|
||||
PackageType: models.PackageGeneric,
|
||||
RepoType: models.RepoTypeRemote,
|
||||
BaseURL: upB.URL,
|
||||
})
|
||||
res, err := testEngine.Fetch(ctx, rOK, "solo.bin", p)
|
||||
if err != nil {
|
||||
t.Fatalf("single-url fetch: %v", err)
|
||||
}
|
||||
if got := readAll(t, res); got != "solo" {
|
||||
t.Fatalf("single-url body=%q, want solo", got)
|
||||
}
|
||||
|
||||
rDead := seed(t, models.Remote{
|
||||
Name: "eng-solo-dead",
|
||||
PackageType: models.PackageGeneric,
|
||||
RepoType: models.RepoTypeRemote,
|
||||
BaseURL: "http://127.0.0.1:1",
|
||||
})
|
||||
if _, err := testEngine.Fetch(ctx, rDead, "x.bin", p); err == nil {
|
||||
t.Fatal("single dead upstream should error, not succeed")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user