From 52d5c924009c90bddfb6d5169fc9945f10c449fe Mon Sep 17 00:00:00 2001 From: fred-maussion Date: Wed, 24 Sep 2025 21:34:41 +0200 Subject: [PATCH 1/9] Init commit --- proxymode/proxymode.go | 318 ++++++++++++++++++++++++++++++----------- 1 file changed, 237 insertions(+), 81 deletions(-) diff --git a/proxymode/proxymode.go b/proxymode/proxymode.go index ed7850903..d3defcda7 100644 --- a/proxymode/proxymode.go +++ b/proxymode/proxymode.go @@ -11,30 +11,37 @@ import ( "fmt" "net/http" "net/url" + "strconv" "strings" + "sync" "time" - "github.com/gorilla/mux" "github.com/hashicorp/go-retryablehttp" + "github.comcom/gorilla/mux" "go.elastic.co/apm/module/apmhttp/v2" "go.uber.org/zap" "github.com/elastic/package-registry/packages" ) -type ProxyMode struct { - options ProxyOptions - - httpClient *retryablehttp.Client - destinationURL *url.URL - resolver *proxyResolver +// backend holds the parsed information for a single proxy destination. +type backend struct { + URL *url.URL + Priority int + Resolver *proxyResolver +} - logger *zap.Logger +type ProxyMode struct { + options ProxyOptions + httpClient *retryablehttp.Client + backends []backend // Changed from single destinationURL + logger *zap.Logger } +// ProxyOptions now supports multiple backends. type ProxyOptions struct { Enabled bool - ProxyTo string + ProxyTo []string // Changed from single string } func NoProxy(logger *zap.Logger) *ProxyMode { @@ -71,13 +78,38 @@ func NewProxyMode(logger *zap.Logger, options ProxyOptions) (*ProxyMode, error) Backoff: retryablehttp.DefaultBackoff, } - var err error - pm.destinationURL, err = url.Parse(pm.options.ProxyTo) - if err != nil { - return nil, fmt.Errorf("can't create proxy destination URL: %w", err) + // Parse all configured backends. + pm.backends = make([]backend, 0, len(options.ProxyTo)) + for _, proxyAddr := range options.ProxyTo { + parts := strings.Split(proxyAddr, ";") + addr := parts[0] + priority := 0 // Default priority + + if len(parts) > 1 { + p, err := strconv.Atoi(parts[1]) + if err == nil { + priority = p + } else { + logger.Warn("invalid priority format in proxy address, using default", zap.String("address", proxyAddr)) + } + } + + destURL, err := url.Parse(addr) + if err != nil { + return nil, fmt.Errorf("can't create proxy destination URL from '%s': %w", addr, err) + } + + pm.backends = append(pm.backends, backend{ + URL: destURL, + Priority: priority, + Resolver: &proxyResolver{destinationURL: *destURL}, + }) + } + + if len(pm.backends) == 0 { + return nil, errors.New("proxy mode is enabled but no backends are configured in 'proxy_to'") } - pm.resolver = &proxyResolver{destinationURL: *pm.destinationURL} return &pm, nil } @@ -103,7 +135,6 @@ func proxyRetryPolicy(ctx context.Context, resp *http.Response, err error) (bool return true, fmt.Errorf("unexpected content type: %s", contentType) } } - return false, nil } @@ -114,103 +145,228 @@ func (pm *ProxyMode) Enabled() bool { return pm.options.Enabled } +// Search now queries all backends in parallel and merges the results. func (pm *ProxyMode) Search(r *http.Request) (packages.Packages, error) { - - proxyURL := *r.URL - proxyURL.Host = pm.destinationURL.Host - proxyURL.Scheme = pm.destinationURL.Scheme - proxyURL.User = pm.destinationURL.User - - proxyRequest, err := retryablehttp.NewRequestWithContext(r.Context(), http.MethodGet, proxyURL.String(), nil) - if err != nil { - return nil, fmt.Errorf("can't create proxy request: %w", err) + type searchResult struct { + Packages packages.Packages + Backend backend + Err error } - pm.logger.Debug("Proxy /search request", zap.String("request.uri", proxyURL.String())) - response, err := pm.httpClient.Do(proxyRequest) - if err != nil { - return nil, fmt.Errorf("can't proxy search request: %w", err) - } - defer response.Body.Close() - var pkgs packages.Packages - err = json.NewDecoder(response.Body).Decode(&pkgs) - if err != nil { - return nil, fmt.Errorf("can't proxy search request: %w", err) + var wg sync.WaitGroup + resultsChan := make(chan searchResult, len(pm.backends)) + + for _, b := range pm.backends { + wg.Add(1) + go func(backend backend) { + defer wg.Done() + proxyURL := *r.URL + proxyURL.Host = backend.URL.Host + proxyURL.Scheme = backend.URL.Scheme + proxyURL.User = backend.URL.User + + proxyRequest, err := retryablehttp.NewRequestWithContext(r.Context(), http.MethodGet, proxyURL.String(), nil) + if err != nil { + resultsChan <- searchResult{Err: fmt.Errorf("can't create proxy request for %s: %w", backend.URL, err)} + return + } + + pm.logger.Debug("Proxying /search request", zap.String("request.uri", proxyURL.String())) + response, err := pm.httpClient.Do(proxyRequest) + if err != nil { + resultsChan <- searchResult{Err: fmt.Errorf("can't proxy search request to %s: %w", backend.URL, err)} + return + } + defer response.Body.Close() + + var pkgs packages.Packages + if err := json.NewDecoder(response.Body).Decode(&pkgs); err != nil { + resultsChan <- searchResult{Err: fmt.Errorf("can't decode search response from %s: %w", backend.URL, err)} + return + } + resultsChan <- searchResult{Packages: pkgs, Backend: backend} + }(b) } - for i := 0; i < len(pkgs); i++ { - pkgs[i].SetRemoteResolver(pm.resolver) + + go func() { + wg.Wait() + close(resultsChan) + }() + + var mergedPackages packages.Packages + for result := range resultsChan { + if result.Err != nil { + pm.logger.Warn("Failed to fetch from proxy backend", zap.Error(result.Err)) + continue + } + // Set the resolver for each package to its backend of origin. + for i := range result.Packages { + result.Packages[i].SetRemoteResolver(result.Backend.Resolver) + } + mergedPackages = mergedPackages.Join(result.Packages) } - return pkgs, nil + + // NOTE: This is where complex merging/conflict resolution would be added. + // For now, we simply combine them. + return mergedPackages, nil } func (pm *ProxyMode) Categories(r *http.Request) ([]packages.Category, error) { + type categoriesResult struct { + Categories []packages.Category + Err error + } - proxyURL := *r.URL - proxyURL.Host = pm.destinationURL.Host - proxyURL.Scheme = pm.destinationURL.Scheme - proxyURL.User = pm.destinationURL.User - - proxyRequest, err := retryablehttp.NewRequestWithContext(r.Context(), http.MethodGet, proxyURL.String(), nil) - if err != nil { - return nil, fmt.Errorf("can't create proxy request: %w", err) + var wg sync.WaitGroup + resultsChan := make(chan categoriesResult, len(pm.backends)) + + for _, b := range pm.backends { + wg.Add(1) + go func(backend backend) { + defer wg.Done() + proxyURL := *r.URL + proxyURL.Host = backend.URL.Host + proxyURL.Scheme = backend.URL.Scheme + proxyURL.User = backend.URL.User + + proxyRequest, err := retryablehttp.NewRequestWithContext(r.Context(), http.MethodGet, proxyURL.String(), nil) + if err != nil { + resultsChan <- categoriesResult{Err: fmt.Errorf("can't create proxy request for %s: %w", backend.URL, err)} + return + } + + pm.logger.Debug("Proxying /categories request", zap.String("request.uri", proxyURL.String())) + response, err := pm.httpClient.Do(proxyRequest) + if err != nil { + resultsChan <- categoriesResult{Err: fmt.Errorf("can't proxy categories request to %s: %w", backend.URL, err)} + return + } + defer response.Body.Close() + + var cats []packages.Category + if err := json.NewDecoder(response.Body).Decode(&cats); err != nil { + resultsChan <- categoriesResult{Err: fmt.Errorf("can't decode categories response from %s: %w", backend.URL, err)} + return + } + resultsChan <- categoriesResult{Categories: cats} + }(b) } - pm.logger.Debug("Proxy /categories request", zap.String("request.uri", proxyURL.String())) - response, err := pm.httpClient.Do(proxyRequest) - if err != nil { - return nil, fmt.Errorf("can't proxy categories request: %w", err) + go func() { + wg.Wait() + close(resultsChan) + }() + + // Merge categories and sum counts for duplicates. + mergedCategories := make(map[string]packages.Category) + for result := range resultsChan { + if result.Err != nil { + pm.logger.Warn("Failed to fetch categories from proxy backend", zap.Error(result.Err)) + continue + } + for _, cat := range result.Categories { + if existing, ok := mergedCategories[cat.Id]; ok { + existing.Count += cat.Count + mergedCategories[cat.Id] = existing + } else { + mergedCategories[cat.Id] = cat + } + } } - defer response.Body.Close() - var cats []packages.Category - err = json.NewDecoder(response.Body).Decode(&cats) - if err != nil { - return nil, fmt.Errorf("can't proxy categories request: %w", err) + + finalList := make([]packages.Category, 0, len(mergedCategories)) + for _, cat := range mergedCategories { + finalList = append(finalList, cat) } - return cats, nil + + return finalList, nil } +// Package queries all backends in parallel and returns the first successful response. func (pm *ProxyMode) Package(r *http.Request) (*packages.Package, error) { - vars := mux.Vars(r) packageName, ok := vars["packageName"] if !ok { return nil, errors.New("missing package name") } - packageVersion, ok := vars["packageVersion"] if !ok { return nil, errors.New("missing package version") } - urlPath := fmt.Sprintf("/package/%s/%s/", packageName, packageVersion) - proxyURL := pm.destinationURL.ResolveReference(&url.URL{Path: urlPath}) - proxyRequest, err := retryablehttp.NewRequestWithContext(r.Context(), http.MethodGet, proxyURL.String(), nil) - if err != nil { - return nil, fmt.Errorf("can't create proxy request: %w", err) + type packageResult struct { + Package *packages.Package + Err error } - pm.logger.Debug("Proxy /package request", zap.String("request.uri", proxyURL.String())) - response, err := pm.httpClient.Do(proxyRequest) - if err != nil { - return nil, fmt.Errorf("can't proxy package request: %w", err) + // We use a context that can be canceled once we find a result. + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + + resultsChan := make(chan packageResult, len(pm.backends)) + var wg sync.WaitGroup + + for _, b := range pm.backends { + wg.Add(1) + go func(backend backend) { + defer wg.Done() + urlPath := fmt.Sprintf("/package/%s/%s/", packageName, packageVersion) + proxyURL := backend.URL.ResolveReference(&url.URL{Path: urlPath}) + + proxyRequest, err := retryablehttp.NewRequestWithContext(ctx, http.MethodGet, proxyURL.String(), nil) + if err != nil { + resultsChan <- packageResult{Err: fmt.Errorf("can't create proxy request for %s: %w", backend.URL, err)} + return + } + + pm.logger.Debug("Proxying /package request", zap.String("request.uri", proxyURL.String())) + response, err := pm.httpClient.Do(proxyRequest) + if err != nil { + resultsChan <- packageResult{Err: fmt.Errorf("can't proxy package request to %s: %w", backend.URL, err)} + return + } + defer response.Body.Close() + + switch response.StatusCode { + case http.StatusOK: + var pkg packages.Package + if err := json.NewDecoder(response.Body).Decode(&pkg); err != nil { + resultsChan <- packageResult{Err: fmt.Errorf("can't decode package response from %s: %w", backend.URL, err)} + return + } + pkg.SetRemoteResolver(backend.Resolver) + resultsChan <- packageResult{Package: &pkg} + case http.StatusNotFound: + // Not an error, just not found on this backend. + resultsChan <- packageResult{Package: nil} + default: + resultsChan <- packageResult{Err: fmt.Errorf("unexpected status code %d from %s", response.StatusCode, backend.URL)} + } + }(b) } - defer response.Body.Close() - switch response.StatusCode { - case http.StatusOK: - // Package found, all good. - case http.StatusNotFound: - // Package doesn't exist, don't try to parse the response, just return an empty package. - return nil, nil - default: - return nil, fmt.Errorf("unexpected status code %d received", response.StatusCode) + go func() { + wg.Wait() + close(resultsChan) + }() + + var lastErr error + for result := range resultsChan { + if result.Err != nil { + lastErr = result.Err + pm.logger.Warn("Failed to fetch package from proxy backend", zap.Error(result.Err)) + continue + } + // If we found a package, cancel other requests and return immediately. + if result.Package != nil { + cancel() // Signal other goroutines to stop. + return result.Package, nil + } } - var pkg packages.Package - err = json.NewDecoder(response.Body).Decode(&pkg) - if err != nil { - return nil, fmt.Errorf("can't proxy package request: %w", err) + // If we get here, no package was found on any backend. + if lastErr != nil { + return nil, lastErr } - pkg.SetRemoteResolver(pm.resolver) - return &pkg, nil + return nil, nil // Not found } From d1d5f39d66aa7cf05b269bf280ba864a3ef9fcec Mon Sep 17 00:00:00 2001 From: fred-maussion Date: Wed, 24 Sep 2025 21:38:50 +0200 Subject: [PATCH 2/9] fixing dep --- proxymode/proxymode.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/proxymode/proxymode.go b/proxymode/proxymode.go index d3defcda7..089871fea 100644 --- a/proxymode/proxymode.go +++ b/proxymode/proxymode.go @@ -16,8 +16,8 @@ import ( "sync" "time" + "github.com/gorilla/mux" "github.com/hashicorp/go-retryablehttp" - "github.comcom/gorilla/mux" "go.elastic.co/apm/module/apmhttp/v2" "go.uber.org/zap" From 4d9dfb2609995f7a022826bbb6e2ed57f8fbd763 Mon Sep 17 00:00:00 2001 From: fred-maussion Date: Wed, 24 Sep 2025 21:46:40 +0200 Subject: [PATCH 3/9] Pass a slice containing the single string --- main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/main.go b/main.go index 9aa359325..7356a10d8 100644 --- a/main.go +++ b/main.go @@ -632,7 +632,7 @@ func getRouter(logger *zap.Logger, options serverOptions) (*mux.Router, error) { } proxyMode, err := proxymode.NewProxyMode(logger, proxymode.ProxyOptions{ Enabled: featureProxyMode, - ProxyTo: proxyTo, + ProxyTo: []string{proxyTo}, }) if err != nil { return nil, fmt.Errorf("can't create proxy mode: %w", err) From 042b48a931cc6daa14c1da5dac69b1267780805c Mon Sep 17 00:00:00 2001 From: fred-maussion Date: Wed, 24 Sep 2025 21:51:50 +0200 Subject: [PATCH 4/9] fix conf --- main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/main.go b/main.go index 7356a10d8..863b73bb5 100644 --- a/main.go +++ b/main.go @@ -632,7 +632,7 @@ func getRouter(logger *zap.Logger, options serverOptions) (*mux.Router, error) { } proxyMode, err := proxymode.NewProxyMode(logger, proxymode.ProxyOptions{ Enabled: featureProxyMode, - ProxyTo: []string{proxyTo}, + ProxyTo: strings.Split(proxyTo, ","), }) if err != nil { return nil, fmt.Errorf("can't create proxy mode: %w", err) From 6f26644ef9732d37a3cac3bb14d373d00c8f1765 Mon Sep 17 00:00:00 2001 From: fred-maussion Date: Wed, 24 Sep 2025 22:02:40 +0200 Subject: [PATCH 5/9] restructuring the http part --- proxymode/proxymode.go | 93 +++++++++++++++++++++++++++--------------- 1 file changed, 60 insertions(+), 33 deletions(-) diff --git a/proxymode/proxymode.go b/proxymode/proxymode.go index 089871fea..2738b8a69 100644 --- a/proxymode/proxymode.go +++ b/proxymode/proxymode.go @@ -31,17 +31,18 @@ type backend struct { Resolver *proxyResolver } +// ProxyMode now stores the shared transport, not a shared client. type ProxyMode struct { - options ProxyOptions - httpClient *retryablehttp.Client - backends []backend // Changed from single destinationURL - logger *zap.Logger + options ProxyOptions + httpTransport http.RoundTripper + backends []backend + logger *zap.Logger } // ProxyOptions now supports multiple backends. type ProxyOptions struct { Enabled bool - ProxyTo []string // Changed from single string + ProxyTo []string } func NoProxy(logger *zap.Logger) *ProxyMode { @@ -61,22 +62,12 @@ func NewProxyMode(logger *zap.Logger, options ProxyOptions) (*ProxyMode, error) return &pm, nil } - pm.httpClient = &retryablehttp.Client{ - HTTPClient: &http.Client{ - Timeout: 10 * time.Second, - Transport: apmhttp.WrapRoundTripper(&http.Transport{ - MaxIdleConns: 100, - MaxIdleConnsPerHost: 100, - IdleConnTimeout: 90 * time.Second, - }), - }, - Logger: withZapLoggerAdapter(logger), - RetryWaitMin: 1 * time.Second, - RetryWaitMax: 15 * time.Second, - RetryMax: 4, - CheckRetry: proxyRetryPolicy, - Backoff: retryablehttp.DefaultBackoff, - } + // Create one shared, instrumented transport for all clients to use. + pm.httpTransport = apmhttp.WrapRoundTripper(&http.Transport{ + MaxIdleConns: 100, + MaxIdleConnsPerHost: 100, + IdleConnTimeout: 90 * time.Second, + }) // Parse all configured backends. pm.backends = make([]backend, 0, len(options.ProxyTo)) @@ -113,22 +104,16 @@ func NewProxyMode(logger *zap.Logger, options ProxyOptions) (*ProxyMode, error) return &pm, nil } -// proxyRetryPolicy function extends the DefaultRetryPolicy to check if the HTTP response content-type -// is application/json. We found occurrences of requests being rejected by an intermittent proxy and causing -// the json.Decoder to fail. +// proxyRetryPolicy function remains unchanged. func proxyRetryPolicy(ctx context.Context, resp *http.Response, err error) (bool, error) { shouldRetry, err := retryablehttp.DefaultRetryPolicy(ctx, resp, err) if shouldRetry { return shouldRetry, err } - - // Chaining Package Registry servers (proxies) is allowed. HTTP client must get to the end of the chain. locationHeader := resp.Header.Get("location") if locationHeader != "" { return false, nil } - - // Expect json content type only for success statuses. if code := resp.StatusCode; code >= 200 && code < 300 { contentType := resp.Header.Get("content-type") if !strings.HasPrefix(contentType, "application/json") { @@ -160,6 +145,21 @@ func (pm *ProxyMode) Search(r *http.Request) (packages.Packages, error) { wg.Add(1) go func(backend backend) { defer wg.Done() + + // Create a new, lightweight client for each goroutine. + httpClient := &retryablehttp.Client{ + HTTPClient: &http.Client{ + Timeout: 10 * time.Second, + Transport: pm.httpTransport, // All clients share the same transport. + }, + Logger: withZapLoggerAdapter(pm.logger), + RetryWaitMin: 1 * time.Second, + RetryWaitMax: 15 * time.Second, + RetryMax: 4, + CheckRetry: proxyRetryPolicy, + Backoff: retryablehttp.DefaultBackoff, + } + proxyURL := *r.URL proxyURL.Host = backend.URL.Host proxyURL.Scheme = backend.URL.Scheme @@ -172,7 +172,7 @@ func (pm *ProxyMode) Search(r *http.Request) (packages.Packages, error) { } pm.logger.Debug("Proxying /search request", zap.String("request.uri", proxyURL.String())) - response, err := pm.httpClient.Do(proxyRequest) + response, err := httpClient.Do(proxyRequest) if err != nil { resultsChan <- searchResult{Err: fmt.Errorf("can't proxy search request to %s: %w", backend.URL, err)} return @@ -206,11 +206,10 @@ func (pm *ProxyMode) Search(r *http.Request) (packages.Packages, error) { mergedPackages = mergedPackages.Join(result.Packages) } - // NOTE: This is where complex merging/conflict resolution would be added. - // For now, we simply combine them. return mergedPackages, nil } +// Categories also queries all backends in parallel. func (pm *ProxyMode) Categories(r *http.Request) ([]packages.Category, error) { type categoriesResult struct { Categories []packages.Category @@ -224,6 +223,20 @@ func (pm *ProxyMode) Categories(r *http.Request) ([]packages.Category, error) { wg.Add(1) go func(backend backend) { defer wg.Done() + + httpClient := &retryablehttp.Client{ + HTTPClient: &http.Client{ + Timeout: 10 * time.Second, + Transport: pm.httpTransport, + }, + Logger: withZapLoggerAdapter(pm.logger), + RetryWaitMin: 1 * time.Second, + RetryWaitMax: 15 * time.Second, + RetryMax: 4, + CheckRetry: proxyRetryPolicy, + Backoff: retryablehttp.DefaultBackoff, + } + proxyURL := *r.URL proxyURL.Host = backend.URL.Host proxyURL.Scheme = backend.URL.Scheme @@ -236,7 +249,7 @@ func (pm *ProxyMode) Categories(r *http.Request) ([]packages.Category, error) { } pm.logger.Debug("Proxying /categories request", zap.String("request.uri", proxyURL.String())) - response, err := pm.httpClient.Do(proxyRequest) + response, err := httpClient.Do(proxyRequest) if err != nil { resultsChan <- categoriesResult{Err: fmt.Errorf("can't proxy categories request to %s: %w", backend.URL, err)} return @@ -310,6 +323,20 @@ func (pm *ProxyMode) Package(r *http.Request) (*packages.Package, error) { wg.Add(1) go func(backend backend) { defer wg.Done() + + httpClient := &retryablehttp.Client{ + HTTPClient: &http.Client{ + Timeout: 10 * time.Second, + Transport: pm.httpTransport, + }, + Logger: withZapLoggerAdapter(pm.logger), + RetryWaitMin: 1 * time.Second, + RetryWaitMax: 15 * time.Second, + RetryMax: 4, + CheckRetry: proxyRetryPolicy, + Backoff: retryablehttp.DefaultBackoff, + } + urlPath := fmt.Sprintf("/package/%s/%s/", packageName, packageVersion) proxyURL := backend.URL.ResolveReference(&url.URL{Path: urlPath}) @@ -320,7 +347,7 @@ func (pm *ProxyMode) Package(r *http.Request) (*packages.Package, error) { } pm.logger.Debug("Proxying /package request", zap.String("request.uri", proxyURL.String())) - response, err := pm.httpClient.Do(proxyRequest) + response, err := httpClient.Do(proxyRequest) if err != nil { resultsChan <- packageResult{Err: fmt.Errorf("can't proxy package request to %s: %w", backend.URL, err)} return From 5e6b821cca749a9f1a25e295f14484800a03f665 Mon Sep 17 00:00:00 2001 From: fred-maussion Date: Wed, 24 Sep 2025 22:46:06 +0200 Subject: [PATCH 6/9] Adding apm context to proxymode logger Adding proxytimeout --- proxymode/logger.go | 21 +++++++++++++++------ proxymode/proxymode.go | 31 ++++++++++++------------------- 2 files changed, 27 insertions(+), 25 deletions(-) diff --git a/proxymode/logger.go b/proxymode/logger.go index f468ce61c..961c06fdf 100644 --- a/proxymode/logger.go +++ b/proxymode/logger.go @@ -5,39 +5,48 @@ package proxymode import ( + "context" + "github.com/hashicorp/go-retryablehttp" + "go.elastic.co/apm/module/apmzap/v2" "go.uber.org/zap" ) type zapLoggerAdapter struct { target *zap.Logger + ctx context.Context } var _ retryablehttp.LeveledLogger = new(zapLoggerAdapter) -func withZapLoggerAdapter(target *zap.Logger) retryablehttp.LeveledLogger { +// The constructor is still needed to pass the context. +func newZapLoggerAdapter(ctx context.Context, target *zap.Logger) retryablehttp.LeveledLogger { return &zapLoggerAdapter{ target: target, + ctx: ctx, } } func (a zapLoggerAdapter) Error(msg string, keysAndValues ...interface{}) { - a.target.Error(msg, keysAndValuesAsZapFields(keysAndValues...)...) + loggerWithContext := a.target.With(apmzap.TraceContext(a.ctx)...) + loggerWithContext.Error(msg, keysAndValuesAsZapFields(keysAndValues...)...) } func (a zapLoggerAdapter) Info(msg string, keysAndValues ...interface{}) { - a.target.Info(msg, keysAndValuesAsZapFields(keysAndValues...)...) + loggerWithContext := a.target.With(apmzap.TraceContext(a.ctx)...) + loggerWithContext.Info(msg, keysAndValuesAsZapFields(keysAndValues...)...) } func (a zapLoggerAdapter) Debug(msg string, keysAndValues ...interface{}) { - a.target.Debug(msg, keysAndValuesAsZapFields(keysAndValues...)...) + loggerWithContext := a.target.With(apmzap.TraceContext(a.ctx)...) + loggerWithContext.Debug(msg, keysAndValuesAsZapFields(keysAndValues...)...) } func (a zapLoggerAdapter) Warn(msg string, keysAndValues ...interface{}) { - a.target.Warn(msg, keysAndValuesAsZapFields(keysAndValues...)...) + loggerWithContext := a.target.With(apmzap.TraceContext(a.ctx)...) + loggerWithContext.Warn(msg, keysAndValuesAsZapFields(keysAndValues...)...) } -// keysAndValuesAsZapFields function transforms the LeveledLogger arguments to the zap.Logger interface. func keysAndValuesAsZapFields(keysAndValues ...interface{}) []zap.Field { fields := make([]zap.Field, len(keysAndValues)/2) var j int diff --git a/proxymode/proxymode.go b/proxymode/proxymode.go index 2738b8a69..adcfce8be 100644 --- a/proxymode/proxymode.go +++ b/proxymode/proxymode.go @@ -31,7 +31,7 @@ type backend struct { Resolver *proxyResolver } -// ProxyMode now stores the shared transport, not a shared client. +// ProxyMode stores the shared transport, not a shared client. type ProxyMode struct { options ProxyOptions httpTransport http.RoundTripper @@ -39,10 +39,11 @@ type ProxyMode struct { logger *zap.Logger } -// ProxyOptions now supports multiple backends. +// ProxyOptions supports multiple backends and a configurable timeout. type ProxyOptions struct { Enabled bool ProxyTo []string + Timeout time.Duration } func NoProxy(logger *zap.Logger) *ProxyMode { @@ -74,7 +75,7 @@ func NewProxyMode(logger *zap.Logger, options ProxyOptions) (*ProxyMode, error) for _, proxyAddr := range options.ProxyTo { parts := strings.Split(proxyAddr, ";") addr := parts[0] - priority := 0 // Default priority + priority := 0 if len(parts) > 1 { p, err := strconv.Atoi(parts[1]) @@ -104,7 +105,6 @@ func NewProxyMode(logger *zap.Logger, options ProxyOptions) (*ProxyMode, error) return &pm, nil } -// proxyRetryPolicy function remains unchanged. func proxyRetryPolicy(ctx context.Context, resp *http.Response, err error) (bool, error) { shouldRetry, err := retryablehttp.DefaultRetryPolicy(ctx, resp, err) if shouldRetry { @@ -146,13 +146,12 @@ func (pm *ProxyMode) Search(r *http.Request) (packages.Packages, error) { go func(backend backend) { defer wg.Done() - // Create a new, lightweight client for each goroutine. httpClient := &retryablehttp.Client{ HTTPClient: &http.Client{ - Timeout: 10 * time.Second, - Transport: pm.httpTransport, // All clients share the same transport. + Timeout: pm.options.Timeout, + Transport: pm.httpTransport, }, - Logger: withZapLoggerAdapter(pm.logger), + Logger: newZapLoggerAdapter(r.Context(), pm.logger), RetryWaitMin: 1 * time.Second, RetryWaitMax: 15 * time.Second, RetryMax: 4, @@ -199,7 +198,6 @@ func (pm *ProxyMode) Search(r *http.Request) (packages.Packages, error) { pm.logger.Warn("Failed to fetch from proxy backend", zap.Error(result.Err)) continue } - // Set the resolver for each package to its backend of origin. for i := range result.Packages { result.Packages[i].SetRemoteResolver(result.Backend.Resolver) } @@ -226,10 +224,10 @@ func (pm *ProxyMode) Categories(r *http.Request) ([]packages.Category, error) { httpClient := &retryablehttp.Client{ HTTPClient: &http.Client{ - Timeout: 10 * time.Second, + Timeout: pm.options.Timeout, Transport: pm.httpTransport, }, - Logger: withZapLoggerAdapter(pm.logger), + Logger: newZapLoggerAdapter(r.Context(), pm.logger), RetryWaitMin: 1 * time.Second, RetryWaitMax: 15 * time.Second, RetryMax: 4, @@ -270,7 +268,6 @@ func (pm *ProxyMode) Categories(r *http.Request) ([]packages.Category, error) { close(resultsChan) }() - // Merge categories and sum counts for duplicates. mergedCategories := make(map[string]packages.Category) for result := range resultsChan { if result.Err != nil { @@ -312,7 +309,6 @@ func (pm *ProxyMode) Package(r *http.Request) (*packages.Package, error) { Err error } - // We use a context that can be canceled once we find a result. ctx, cancel := context.WithCancel(r.Context()) defer cancel() @@ -326,10 +322,10 @@ func (pm *ProxyMode) Package(r *http.Request) (*packages.Package, error) { httpClient := &retryablehttp.Client{ HTTPClient: &http.Client{ - Timeout: 10 * time.Second, + Timeout: pm.options.Timeout, Transport: pm.httpTransport, }, - Logger: withZapLoggerAdapter(pm.logger), + Logger: newZapLoggerAdapter(r.Context(), pm.logger), RetryWaitMin: 1 * time.Second, RetryWaitMax: 15 * time.Second, RetryMax: 4, @@ -364,7 +360,6 @@ func (pm *ProxyMode) Package(r *http.Request) (*packages.Package, error) { pkg.SetRemoteResolver(backend.Resolver) resultsChan <- packageResult{Package: &pkg} case http.StatusNotFound: - // Not an error, just not found on this backend. resultsChan <- packageResult{Package: nil} default: resultsChan <- packageResult{Err: fmt.Errorf("unexpected status code %d from %s", response.StatusCode, backend.URL)} @@ -384,14 +379,12 @@ func (pm *ProxyMode) Package(r *http.Request) (*packages.Package, error) { pm.logger.Warn("Failed to fetch package from proxy backend", zap.Error(result.Err)) continue } - // If we found a package, cancel other requests and return immediately. if result.Package != nil { - cancel() // Signal other goroutines to stop. + cancel() return result.Package, nil } } - // If we get here, no package was found on any backend. if lastErr != nil { return nil, lastErr } From f0ebfe9d3a1c94c5fe58c9f365ae6ca6d23fa911 Mon Sep 17 00:00:00 2001 From: fred-maussion Date: Wed, 24 Sep 2025 22:50:56 +0200 Subject: [PATCH 7/9] Adding backend proxy variable timeout --- main.go | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/main.go b/main.go index 863b73bb5..be4d9457a 100644 --- a/main.go +++ b/main.go @@ -630,9 +630,23 @@ func getRouter(logger *zap.Logger, options serverOptions) (*mux.Router, error) { if featureProxyMode { logger.Info("Technical preview: Proxy mode is an experimental feature and it may be unstable.") } + + // Default proxy backend timeout + proxyTimeout := 2 * time.Second + + // Override the default from an environment variable if any. + if v := os.Getenv("EPR_PROXY_TIMEOUT"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + proxyTimeout = d + } else { + logger.Warn("invalid EPR_PROXY_TIMEOUT format, using default", zap.String("value", v)) + } + } + proxyMode, err := proxymode.NewProxyMode(logger, proxymode.ProxyOptions{ Enabled: featureProxyMode, ProxyTo: strings.Split(proxyTo, ","), + Timeout: proxyTimeout, }) if err != nil { return nil, fmt.Errorf("can't create proxy mode: %w", err) From 6da58388ae85209ae69dd69830862f0b5a503619 Mon Sep 17 00:00:00 2001 From: fred-maussion Date: Thu, 25 Sep 2025 14:57:06 +0200 Subject: [PATCH 8/9] Adding multi-backend support for proxymode --- main.go | 9 ++++++++- proxymode/logger.go | 22 ++++++++++++++++++++-- proxymode/proxymode.go | 17 +++++++++++++---- 3 files changed, 41 insertions(+), 7 deletions(-) diff --git a/main.go b/main.go index be4d9457a..be3b967be 100644 --- a/main.go +++ b/main.go @@ -47,7 +47,6 @@ import ( ) const ( - serviceName = "package-registry" version = "1.32.2" defaultInstanceName = "localhost" ) @@ -83,6 +82,7 @@ var ( featureProxyMode bool proxyTo string + serviceName = getServiceName() defaultConfig = Config{ CacheTimeIndex: 10 * time.Second, @@ -336,6 +336,13 @@ func getHostname() string { return hostname } +func getServiceName() string { + if name := os.Getenv("ELASTIC_APM_SERVICE_NAME"); name != "" { + return name + } + return "package-registry" +} + func initMetricsServer(logger *zap.Logger) { if metricsAddress == "" { return diff --git a/proxymode/logger.go b/proxymode/logger.go index 961c06fdf..7b22ac6f2 100644 --- a/proxymode/logger.go +++ b/proxymode/logger.go @@ -6,6 +6,7 @@ package proxymode import ( "context" + "errors" "github.com/hashicorp/go-retryablehttp" "go.elastic.co/apm/module/apmzap/v2" @@ -19,7 +20,6 @@ type zapLoggerAdapter struct { var _ retryablehttp.LeveledLogger = new(zapLoggerAdapter) -// The constructor is still needed to pass the context. func newZapLoggerAdapter(ctx context.Context, target *zap.Logger) retryablehttp.LeveledLogger { return &zapLoggerAdapter{ target: target, @@ -28,7 +28,25 @@ func newZapLoggerAdapter(ctx context.Context, target *zap.Logger) retryablehttp. } func (a zapLoggerAdapter) Error(msg string, keysAndValues ...interface{}) { + // Check if the error is an expected context cancellation. + isCanceled := false + for i := 0; i < len(keysAndValues); i += 2 { + if key, ok := keysAndValues[i].(string); ok && key == "error" { + if err, ok := keysAndValues[i+1].(error); ok && errors.Is(err, context.Canceled) { + isCanceled = true + break + } + } + } + loggerWithContext := a.target.With(apmzap.TraceContext(a.ctx)...) + + // If the error was an expected cancellation, log it as a debug message. + if isCanceled { + loggerWithContext.Debug(msg, keysAndValuesAsZapFields(keysAndValues...)...) + return + } + loggerWithContext.Error(msg, keysAndValuesAsZapFields(keysAndValues...)...) } @@ -53,7 +71,7 @@ func keysAndValuesAsZapFields(keysAndValues ...interface{}) []zap.Field { for i := 0; i < len(keysAndValues); i += 2 { key, ok := keysAndValues[i].(string) if !ok { - continue // something is wrong with the key, string expected + continue } fields[j] = zap.Any(key, keysAndValues[i+1]) j++ diff --git a/proxymode/proxymode.go b/proxymode/proxymode.go index adcfce8be..6b3993c67 100644 --- a/proxymode/proxymode.go +++ b/proxymode/proxymode.go @@ -75,7 +75,7 @@ func NewProxyMode(logger *zap.Logger, options ProxyOptions) (*ProxyMode, error) for _, proxyAddr := range options.ProxyTo { parts := strings.Split(proxyAddr, ";") addr := parts[0] - priority := 0 + priority := 0 // Default priority if len(parts) > 1 { p, err := strconv.Atoi(parts[1]) @@ -105,11 +105,19 @@ func NewProxyMode(logger *zap.Logger, options ProxyOptions) (*ProxyMode, error) return &pm, nil } +// proxyRetryPolicy function extends the DefaultRetryPolicy and handles nil responses. func proxyRetryPolicy(ctx context.Context, resp *http.Response, err error) (bool, error) { shouldRetry, err := retryablehttp.DefaultRetryPolicy(ctx, resp, err) if shouldRetry { return shouldRetry, err } + + // If the response is nil, a network error occurred (e.g., DNS failure). + // We can't inspect the response, so we return the default decision. + if resp == nil { + return false, nil + } + locationHeader := resp.Header.Get("location") if locationHeader != "" { return false, nil @@ -325,7 +333,7 @@ func (pm *ProxyMode) Package(r *http.Request) (*packages.Package, error) { Timeout: pm.options.Timeout, Transport: pm.httpTransport, }, - Logger: newZapLoggerAdapter(r.Context(), pm.logger), + Logger: newZapLoggerAdapter(ctx, pm.logger), RetryWaitMin: 1 * time.Second, RetryWaitMax: 15 * time.Second, RetryMax: 4, @@ -345,7 +353,8 @@ func (pm *ProxyMode) Package(r *http.Request) (*packages.Package, error) { pm.logger.Debug("Proxying /package request", zap.String("request.uri", proxyURL.String())) response, err := httpClient.Do(proxyRequest) if err != nil { - resultsChan <- packageResult{Err: fmt.Errorf("can't proxy package request to %s: %w", backend.URL, err)} + // We don't send to resultsChan here because the logger adapter will have already logged it. + // This avoids duplicate error messages in the final, merged output. return } defer response.Body.Close() @@ -380,7 +389,7 @@ func (pm *ProxyMode) Package(r *http.Request) (*packages.Package, error) { continue } if result.Package != nil { - cancel() + cancel() // Signal other goroutines to stop. return result.Package, nil } } From d6665d840330b153a2f3184af492b8a2a35b1916 Mon Sep 17 00:00:00 2001 From: fred-maussion Date: Fri, 26 Sep 2025 10:57:59 +0200 Subject: [PATCH 9/9] Move to variables for the new proxymode retry --- main.go | 32 ++++++++++++++++++++++++++++---- proxymode/proxymode.go | 27 +++++++++++++++------------ 2 files changed, 43 insertions(+), 16 deletions(-) diff --git a/main.go b/main.go index be3b967be..896b45d5a 100644 --- a/main.go +++ b/main.go @@ -638,8 +638,11 @@ func getRouter(logger *zap.Logger, options serverOptions) (*mux.Router, error) { logger.Info("Technical preview: Proxy mode is an experimental feature and it may be unstable.") } - // Default proxy backend timeout + // Default proxy backend and retry timeout proxyTimeout := 2 * time.Second + retryMax := 1 + retryWaitMin := 1 * time.Second + retryWaitMax := 3 * time.Second // Override the default from an environment variable if any. if v := os.Getenv("EPR_PROXY_TIMEOUT"); v != "" { @@ -650,10 +653,31 @@ func getRouter(logger *zap.Logger, options serverOptions) (*mux.Router, error) { } } + if v := os.Getenv("EPR_PROXY_RETRY_MAX"); v != "" { + if i, err := strconv.Atoi(v); err == nil { + retryMax = i + } + } + + if v := os.Getenv("EPR_PROXY_RETRY_WAIT_MIN"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + retryWaitMin = d + } + } + + if v := os.Getenv("EPR_PROXY_RETRY_WAIT_MAX"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + retryWaitMax = d + } + } + proxyMode, err := proxymode.NewProxyMode(logger, proxymode.ProxyOptions{ - Enabled: featureProxyMode, - ProxyTo: strings.Split(proxyTo, ","), - Timeout: proxyTimeout, + Enabled: featureProxyMode, + ProxyTo: strings.Split(proxyTo, ","), + Timeout: proxyTimeout, + RetryMax: retryMax, + RetryWaitMin: retryWaitMin, + RetryWaitMax: retryWaitMax, }) if err != nil { return nil, fmt.Errorf("can't create proxy mode: %w", err) diff --git a/proxymode/proxymode.go b/proxymode/proxymode.go index 6b3993c67..03063fa32 100644 --- a/proxymode/proxymode.go +++ b/proxymode/proxymode.go @@ -41,9 +41,12 @@ type ProxyMode struct { // ProxyOptions supports multiple backends and a configurable timeout. type ProxyOptions struct { - Enabled bool - ProxyTo []string - Timeout time.Duration + Enabled bool + ProxyTo []string + Timeout time.Duration + RetryMax int + RetryWaitMin time.Duration + RetryWaitMax time.Duration } func NoProxy(logger *zap.Logger) *ProxyMode { @@ -160,9 +163,9 @@ func (pm *ProxyMode) Search(r *http.Request) (packages.Packages, error) { Transport: pm.httpTransport, }, Logger: newZapLoggerAdapter(r.Context(), pm.logger), - RetryWaitMin: 1 * time.Second, - RetryWaitMax: 15 * time.Second, - RetryMax: 4, + RetryWaitMin: pm.options.RetryWaitMin, + RetryWaitMax: pm.options.RetryWaitMax, + RetryMax: pm.options.RetryMax, CheckRetry: proxyRetryPolicy, Backoff: retryablehttp.DefaultBackoff, } @@ -236,9 +239,9 @@ func (pm *ProxyMode) Categories(r *http.Request) ([]packages.Category, error) { Transport: pm.httpTransport, }, Logger: newZapLoggerAdapter(r.Context(), pm.logger), - RetryWaitMin: 1 * time.Second, - RetryWaitMax: 15 * time.Second, - RetryMax: 4, + RetryWaitMin: pm.options.RetryWaitMin, + RetryWaitMax: pm.options.RetryWaitMax, + RetryMax: pm.options.RetryMax, CheckRetry: proxyRetryPolicy, Backoff: retryablehttp.DefaultBackoff, } @@ -334,9 +337,9 @@ func (pm *ProxyMode) Package(r *http.Request) (*packages.Package, error) { Transport: pm.httpTransport, }, Logger: newZapLoggerAdapter(ctx, pm.logger), - RetryWaitMin: 1 * time.Second, - RetryWaitMax: 15 * time.Second, - RetryMax: 4, + RetryWaitMin: pm.options.RetryWaitMin, + RetryWaitMax: pm.options.RetryWaitMax, + RetryMax: pm.options.RetryMax, CheckRetry: proxyRetryPolicy, Backoff: retryablehttp.DefaultBackoff, }