diff --git a/main.go b/main.go index 9aa359325..896b45d5a 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 @@ -630,9 +637,47 @@ 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 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 != "" { + if d, err := time.ParseDuration(v); err == nil { + proxyTimeout = d + } else { + logger.Warn("invalid EPR_PROXY_TIMEOUT format, using default", zap.String("value", v)) + } + } + + 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: proxyTo, + 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/logger.go b/proxymode/logger.go index f468ce61c..7b22ac6f2 100644 --- a/proxymode/logger.go +++ b/proxymode/logger.go @@ -5,46 +5,73 @@ package proxymode import ( + "context" + "errors" + "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 { +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...)...) + // 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...)...) } 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 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 ed7850903..03063fa32 100644 --- a/proxymode/proxymode.go +++ b/proxymode/proxymode.go @@ -11,7 +11,9 @@ import ( "fmt" "net/http" "net/url" + "strconv" "strings" + "sync" "time" "github.com/gorilla/mux" @@ -22,19 +24,29 @@ import ( "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 +// ProxyMode stores the shared transport, not a shared client. +type ProxyMode struct { + options ProxyOptions + httpTransport http.RoundTripper + backends []backend + logger *zap.Logger } +// ProxyOptions supports multiple backends and a configurable timeout. type ProxyOptions struct { - Enabled bool - ProxyTo string + Enabled bool + ProxyTo []string + Timeout time.Duration + RetryMax int + RetryWaitMin time.Duration + RetryWaitMax time.Duration } func NoProxy(logger *zap.Logger) *ProxyMode { @@ -54,56 +66,71 @@ 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, - } - - 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) + // 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)) + 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 } -// 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 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 } - // Chaining Package Registry servers (proxies) is allowed. HTTP client must get to the end of the chain. + // 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 } - - // 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") { return true, fmt.Errorf("unexpected content type: %s", contentType) } } - return false, nil } @@ -114,103 +141,264 @@ 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() + + httpClient := &retryablehttp.Client{ + HTTPClient: &http.Client{ + Timeout: pm.options.Timeout, + Transport: pm.httpTransport, + }, + Logger: newZapLoggerAdapter(r.Context(), pm.logger), + RetryWaitMin: pm.options.RetryWaitMin, + RetryWaitMax: pm.options.RetryWaitMax, + RetryMax: pm.options.RetryMax, + CheckRetry: proxyRetryPolicy, + Backoff: retryablehttp.DefaultBackoff, + } + + 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 := 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 + } + for i := range result.Packages { + result.Packages[i].SetRemoteResolver(result.Backend.Resolver) + } + mergedPackages = mergedPackages.Join(result.Packages) } - return pkgs, nil + + 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 + 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() + + httpClient := &retryablehttp.Client{ + HTTPClient: &http.Client{ + Timeout: pm.options.Timeout, + Transport: pm.httpTransport, + }, + Logger: newZapLoggerAdapter(r.Context(), pm.logger), + RetryWaitMin: pm.options.RetryWaitMin, + RetryWaitMax: pm.options.RetryWaitMax, + RetryMax: pm.options.RetryMax, + CheckRetry: proxyRetryPolicy, + Backoff: retryablehttp.DefaultBackoff, + } + + 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 := 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) + }() + + 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) + 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() + + httpClient := &retryablehttp.Client{ + HTTPClient: &http.Client{ + Timeout: pm.options.Timeout, + Transport: pm.httpTransport, + }, + Logger: newZapLoggerAdapter(ctx, pm.logger), + RetryWaitMin: pm.options.RetryWaitMin, + RetryWaitMax: pm.options.RetryWaitMax, + RetryMax: pm.options.RetryMax, + CheckRetry: proxyRetryPolicy, + Backoff: retryablehttp.DefaultBackoff, + } + + 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 := httpClient.Do(proxyRequest) + if err != nil { + // 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() + + 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: + 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 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 lastErr != nil { + return nil, lastErr } - pkg.SetRemoteResolver(pm.resolver) - return &pkg, nil + return nil, nil // Not found }