From 5a884ff2a4a789b157ae8bf9cfaf9cad91adf290 Mon Sep 17 00:00:00 2001 From: Denis Makarenko Date: Mon, 21 Sep 2026 11:49:09 +0400 Subject: [PATCH] Opt-in retries for HTTP origin connections --- cmd/cloudflared/tunnel/cmd.go | 6 + cmd/cloudflared/tunnel/origin_retry_test.go | 62 ++++ config/configuration.go | 44 +-- config/duration_test.go | 51 +++ ingress/config.go | 25 +- ingress/ingress.go | 22 +- ingress/origin_connection_test.go | 2 +- ingress/origin_retry.go | 74 +++++ ingress/origin_retry_config_test.go | 99 ++++++ ingress/origin_retry_http_test.go | 221 ++++++++++++ ingress/origin_retry_posix.go | 7 + ingress/origin_retry_posix_test.go | 351 ++++++++++++++++++++ ingress/origin_retry_test.go | 93 ++++++ ingress/origin_retry_windows.go | 5 + ingress/origin_service.go | 37 ++- orchestration/config_test.go | 22 +- 16 files changed, 1072 insertions(+), 49 deletions(-) create mode 100644 cmd/cloudflared/tunnel/origin_retry_test.go create mode 100644 config/duration_test.go create mode 100644 ingress/origin_retry.go create mode 100644 ingress/origin_retry_config_test.go create mode 100644 ingress/origin_retry_http_test.go create mode 100644 ingress/origin_retry_posix.go create mode 100644 ingress/origin_retry_posix_test.go create mode 100644 ingress/origin_retry_test.go create mode 100644 ingress/origin_retry_windows.go diff --git a/cmd/cloudflared/tunnel/cmd.go b/cmd/cloudflared/tunnel/cmd.go index e5b2f4f72fa..44e819d0965 100644 --- a/cmd/cloudflared/tunnel/cmd.go +++ b/cmd/cloudflared/tunnel/cmd.go @@ -975,6 +975,12 @@ func configureProxyFlags(shouldHide bool) []cli.Flag { Value: time.Second * 30, Hidden: shouldHide, }), + altsrc.NewDurationFlag(&cli.DurationFlag{ + Name: ingress.ProxyConnectRetryTimeoutFlag, + Usage: "Total time to retry refused HTTP origin connections or missing Unix sockets (0 disables retries). Applies to --url or --unix-socket; for ingress rules, set originRequest.connectRetryTimeout.", + EnvVars: []string{"TUNNEL_PROXY_CONNECT_RETRY_TIMEOUT"}, + Hidden: shouldHide, + }), altsrc.NewDurationFlag(&cli.DurationFlag{ Name: ingress.ProxyTLSTimeoutFlag, Usage: legacyTunnelFlag("HTTP proxy timeout for completing a TLS handshake"), diff --git a/cmd/cloudflared/tunnel/origin_retry_test.go b/cmd/cloudflared/tunnel/origin_retry_test.go new file mode 100644 index 00000000000..409f7f966fa --- /dev/null +++ b/cmd/cloudflared/tunnel/origin_retry_test.go @@ -0,0 +1,62 @@ +package tunnel + +import ( + "testing" + "time" + + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" + "github.com/urfave/cli/v2" + + "github.com/cloudflare/cloudflared/config" + "github.com/cloudflare/cloudflared/ingress" +) + +func TestOriginConnectRetryFlag(t *testing.T) { + testCases := []struct { + name string + env string + flag string + expected time.Duration + wantErr bool + }{ + {name: "disabled by default"}, + {name: "environment", env: "500ms", expected: 500 * time.Millisecond}, + {name: "flag", flag: "500ms", expected: 500 * time.Millisecond}, + {name: "flag overrides environment", env: "500ms", flag: "2s", expected: 2 * time.Second}, + {name: "flag disables environment", env: "500ms", flag: "0s"}, + {name: "negative flag", flag: "-1s", wantErr: true}, + {name: "negative environment", env: "-1s", wantErr: true}, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Setenv("TUNNEL_PROXY_CONNECT_RETRY_TIMEOUT", tc.env) + for _, origin := range [][]string{{"--url", "http://localhost:8000"}, {"--unix-socket", "/tmp/app.sock"}} { + t.Run(origin[0], func(t *testing.T) { + app := cli.NewApp() + app.Flags = configureProxyFlags(false) + app.Action = func(c *cli.Context) error { + log := zerolog.Nop() + rules, err := ingress.ParseIngressFromConfigAndCLI(&config.Configuration{}, c, &log) + if err != nil { + return err + } + require.Len(t, rules.Rules, 1) + require.Equal(t, tc.expected, rules.Rules[0].Config.ConnectRetryTimeout.Duration) + return nil + } + args := append([]string{"cloudflared"}, origin...) + if tc.flag != "" { + args = append(args, "--proxy-connect-retry-timeout", tc.flag) + } + err := app.Run(args) + if tc.wantErr { + require.ErrorContains(t, err, "connectRetryTimeout must not be negative") + return + } + require.NoError(t, err) + }) + } + }) + } +} diff --git a/config/configuration.go b/config/configuration.go index cb0b0adeda5..0825cd77b9a 100644 --- a/config/configuration.go +++ b/config/configuration.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "io" + "math" "net/url" "os" "path/filepath" @@ -35,7 +36,7 @@ var ( defaultUserConfigDirs = []string{"~/.cloudflared", "~/.cloudflare-warp", "~/cloudflare-warp"} defaultNixConfigDirs = []string{"/etc/cloudflared", DefaultUnixConfigLocation} - ErrNoConfigFile = fmt.Errorf("Cannot determine default configuration path. No file %v in %v", DefaultConfigFiles, DefaultConfigSearchDirectories()) + ErrNoConfigFile = fmt.Errorf("Cannot determine default configuration path. No file %v in %v", DefaultConfigFiles, DefaultConfigSearchDirectories()) //nolint:staticcheck // Preserving the existing user-facing error text. ) const ( @@ -49,7 +50,7 @@ func DefaultConfigDirectory() string { path := os.Getenv("CFDPATH") if path == "" { path = filepath.Join(os.Getenv("ProgramFiles(x86)"), "cloudflared") - if _, err := os.Stat(path); os.IsNotExist(err) { // doesn't exist, so return an empty failure string + if _, err := os.Stat(path); os.IsNotExist(err) { //nolint:gosec // Installation path comes from the local process environment. return "" } } @@ -87,7 +88,7 @@ func DefaultConfigSearchDirectories() []string { // FileExists checks to see if a file exist at the provided path. func FileExists(path string) (bool, error) { - f, err := os.Open(path) + f, err := os.Open(path) //nolint:gosec // Checking a caller-supplied configuration path. if err != nil { if os.IsNotExist(err) { // ignore missing files @@ -126,19 +127,19 @@ func FindOrCreateConfigPath() string { if path == "" { // create the default directory if it doesn't exist path = DefaultConfigPath() - if err := os.MkdirAll(filepath.Dir(path), os.ModePerm); err != nil { + if err := os.MkdirAll(filepath.Dir(path), os.ModePerm); err != nil { //nolint:gosec // Preserving existing umask-controlled directory permissions. return "" } // write a new config file out - file, err := os.Create(path) + file, err := os.Create(path) //nolint:gosec // The path is selected by FindDefaultConfigPath or DefaultConfigPath. if err != nil { return "" } - defer file.Close() + defer func() { _ = file.Close() }() logDir := DefaultLogDirectory() - _ = os.MkdirAll(logDir, os.ModePerm) // try and create it. Doesn't matter if it succeed or not, only byproduct will be no logs + _ = os.MkdirAll(logDir, os.ModePerm) //nolint:gosec // Best-effort log directory creation with existing umask-controlled permissions. c := Root{ LogDirectory: logDir, @@ -187,10 +188,12 @@ type UnvalidatedIngressRule struct { // config. // Note: // - To specify a time.Duration in go-yaml, use e.g. "3s" or "24h". -// - To specify a time.Duration in json, use int64 of the nanoseconds +// - To specify a time.Duration in JSON, use seconds (e.g. 0.5 or 3). type OriginRequestConfig struct { // HTTP proxy timeout for establishing a new connection ConnectTimeout *CustomDuration `yaml:"connectTimeout" json:"connectTimeout,omitempty"` + // Total time to retry refused HTTP origin connections or missing Unix sockets. Zero disables retries. + ConnectRetryTimeout *CustomDuration `yaml:"connectRetryTimeout" json:"connectRetryTimeout,omitempty"` // HTTP proxy timeout for completing a TLS handshake TLSTimeout *CustomDuration `yaml:"tlsTimeout" json:"tlsTimeout,omitempty"` // HTTP proxy TCP keepalive duration @@ -391,7 +394,7 @@ func ReadConfigFile(c *cli.Context, log *zerolog.Logger) (settings *configFileSe } log.Debug().Msgf("Loading configuration from %s", configFile) - file, err := os.Open(configFile) + file, err := os.Open(configFile) //nolint:gosec // Config path is explicitly selected by the local operator. if err != nil { // If does not exist and config file was not specificly specified then return ErrNoConfigFile found. if os.IsNotExist(err) && !c.IsSet("config") { @@ -399,7 +402,7 @@ func ReadConfigFile(c *cli.Context, log *zerolog.Logger) (settings *configFileSe } return nil, "", err } - defer file.Close() + defer func() { _ = file.Close() }() if err := yaml.NewDecoder(file).Decode(&configuration); err != nil { if err == io.EOF { log.Error().Msgf("Configuration file %s was empty", configFile) @@ -410,7 +413,8 @@ func ReadConfigFile(c *cli.Context, log *zerolog.Logger) (settings *configFileSe configuration.sourceFile = configFile // Parse it again, with strict mode, to find warnings. - if file, err := os.Open(configFile); err == nil { + if file, err := os.Open(configFile); err == nil { //nolint:gosec // Re-reading the same operator-selected configuration file. + defer func() { _ = file.Close() }() decoder := yaml.NewDecoder(file) decoder.KnownFields(true) var unusedConfig configFileSettings @@ -422,31 +426,31 @@ func ReadConfigFile(c *cli.Context, log *zerolog.Logger) (settings *configFileSe return &configuration, warnings, nil } -// A CustomDuration is a Duration that has custom serialization for JSON. -// JSON in Javascript assumes that int fields are 32 bits and Duration fields are deserialized assuming that numbers -// are in nanoseconds, which in 32bit integers limits to just 2 seconds. -// This type assumes that when serializing/deserializing from JSON, that the number is in seconds, while it maintains -// the YAML serde assumptions. +// A duration encoded as seconds in JSON and as a Go duration string in YAML. type CustomDuration struct { time.Duration } func (s CustomDuration) MarshalJSON() ([]byte, error) { - return json.Marshal(s.Duration.Seconds()) + return json.Marshal(s.Seconds()) } func (s *CustomDuration) UnmarshalJSON(data []byte) error { - seconds, err := strconv.ParseInt(string(data), 10, 64) + seconds, err := strconv.ParseFloat(string(data), 64) if err != nil { return err } - s.Duration = time.Duration(seconds * int64(time.Second)) + nanoseconds := math.Round(seconds * float64(time.Second)) + if math.IsNaN(nanoseconds) || nanoseconds >= float64(math.MaxInt64) || nanoseconds < float64(math.MinInt64) { + return fmt.Errorf("duration %s seconds is out of range", data) + } + s.Duration = time.Duration(nanoseconds) return nil } func (s *CustomDuration) MarshalYAML() (interface{}, error) { - return s.Duration.String(), nil + return s.String(), nil } func (s *CustomDuration) UnmarshalYAML(unmarshal func(interface{}) error) error { diff --git a/config/duration_test.go b/config/duration_test.go new file mode 100644 index 00000000000..5a059b31167 --- /dev/null +++ b/config/duration_test.go @@ -0,0 +1,51 @@ +package config + +import ( + "encoding/json" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestCustomDurationJSON(t *testing.T) { + t.Parallel() + testCases := []struct { + json string + duration time.Duration + }{ + {"0", 0}, + {"0.000000001", time.Nanosecond}, + {"0.5", 500 * time.Millisecond}, + {"1.001", 1001 * time.Millisecond}, + {"-0.5", -500 * time.Millisecond}, + {"1", time.Second}, + {"30", 30 * time.Second}, + {"3600", time.Hour}, + } + for _, tc := range testCases { + t.Run(tc.json, func(t *testing.T) { + t.Parallel() + var decoded CustomDuration + err := json.Unmarshal([]byte(tc.json), &decoded) + require.NoError(t, err) + require.Equal(t, tc.duration, decoded.Duration) + + data, err := json.Marshal(CustomDuration{Duration: tc.duration}) + require.NoError(t, err) + require.JSONEq(t, tc.json, string(data)) + }) + } +} + +func TestCustomDurationRejectsInvalidJSON(t *testing.T) { + t.Parallel() + for _, data := range []string{`null`, `"1s"`, `1e100`, `-1e100`, `9223372037`, `-9223372037`} { + t.Run(data, func(t *testing.T) { + t.Parallel() + var decoded CustomDuration + err := json.Unmarshal([]byte(data), &decoded) + require.Error(t, err) + }) + } +} diff --git a/ingress/config.go b/ingress/config.go index 83f893fefe1..808977188ad 100644 --- a/ingress/config.go +++ b/ingress/config.go @@ -26,6 +26,7 @@ const ( SSHServerFlag = "ssh-server" Socks5Flag = "socks5" ProxyConnectTimeoutFlag = "proxy-connect-timeout" + ProxyConnectRetryTimeoutFlag = "proxy-connect-retry-timeout" ProxyTLSTimeoutFlag = "proxy-tls-timeout" ProxyTCPKeepAliveFlag = "proxy-tcp-keepalive" ProxyNoHappyEyeballsFlag = "proxy-no-happy-eyeballs" @@ -121,6 +122,7 @@ func (rc *RemoteConfig) UnmarshalJSON(b []byte) error { func originRequestFromSingleRule(c *cli.Context) OriginRequestConfig { var connectTimeout = defaultHTTPConnectTimeout + var connectRetryTimeout config.CustomDuration var tlsTimeout = defaultTLSTimeout var tcpKeepAlive = defaultTCPKeepAlive var noHappyEyeballs bool @@ -140,6 +142,9 @@ func originRequestFromSingleRule(c *cli.Context) OriginRequestConfig { if flag := ProxyConnectTimeoutFlag; c.IsSet(flag) { connectTimeout = config.CustomDuration{Duration: c.Duration(flag)} } + if flag := ProxyConnectRetryTimeoutFlag; c.IsSet(flag) { + connectRetryTimeout = config.CustomDuration{Duration: c.Duration(flag)} + } if flag := ProxyTLSTimeoutFlag; c.IsSet(flag) { tlsTimeout = config.CustomDuration{Duration: c.Duration(flag)} } @@ -193,6 +198,7 @@ func originRequestFromSingleRule(c *cli.Context) OriginRequestConfig { return OriginRequestConfig{ ConnectTimeout: connectTimeout, + ConnectRetryTimeout: connectRetryTimeout, TLSTimeout: tlsTimeout, TCPKeepAlive: tcpKeepAlive, NoHappyEyeballs: noHappyEyeballs, @@ -224,6 +230,9 @@ func originRequestFromConfig(c config.OriginRequestConfig) OriginRequestConfig { if c.ConnectTimeout != nil { out.ConnectTimeout = *c.ConnectTimeout } + if c.ConnectRetryTimeout != nil { + out.ConnectRetryTimeout = *c.ConnectRetryTimeout + } if c.TLSTimeout != nil { out.TLSTimeout = *c.TLSTimeout } @@ -292,6 +301,8 @@ func originRequestFromConfig(c config.OriginRequestConfig) OriginRequestConfig { type OriginRequestConfig struct { // HTTP proxy timeout for establishing a new connection ConnectTimeout config.CustomDuration `yaml:"connectTimeout" json:"connectTimeout"` + // Total time to retry refused HTTP origin connections or missing Unix sockets. Zero disables retries. + ConnectRetryTimeout config.CustomDuration `yaml:"connectRetryTimeout" json:"connectRetryTimeout,omitzero"` // HTTP proxy timeout for completing a TLS handshake TLSTimeout config.CustomDuration `yaml:"tlsTimeout" json:"tlsTimeout"` // HTTP proxy TCP keepalive duration @@ -341,6 +352,12 @@ func (defaults *OriginRequestConfig) setConnectTimeout(overrides config.OriginRe } } +func (defaults *OriginRequestConfig) setConnectRetryTimeout(overrides config.OriginRequestConfig) { + if val := overrides.ConnectRetryTimeout; val != nil { + defaults.ConnectRetryTimeout = *val + } +} + func (defaults *OriginRequestConfig) setTLSTimeout(overrides config.OriginRequestConfig) { if val := overrides.TLSTimeout; val != nil { defaults.TLSTimeout = *val @@ -467,6 +484,7 @@ func (defaults *OriginRequestConfig) setAccess(overrides config.OriginRequestCon func setConfig(defaults OriginRequestConfig, overrides config.OriginRequestConfig) OriginRequestConfig { cfg := defaults cfg.setConnectTimeout(overrides) + cfg.setConnectRetryTimeout(overrides) cfg.setTLSTimeout(overrides) cfg.setNoHappyEyeballs(overrides) cfg.setKeepAliveConnections(overrides) @@ -491,6 +509,7 @@ func setConfig(defaults OriginRequestConfig, overrides config.OriginRequestConfi func ConvertToRawOriginConfig(c OriginRequestConfig) config.OriginRequestConfig { var connectTimeout *config.CustomDuration + var connectRetryTimeout *config.CustomDuration var tlsTimeout *config.CustomDuration var tcpKeepAlive *config.CustomDuration var keepAliveConnections *int @@ -501,6 +520,9 @@ func ConvertToRawOriginConfig(c OriginRequestConfig) config.OriginRequestConfig if c.ConnectTimeout != defaultHTTPConnectTimeout { connectTimeout = &c.ConnectTimeout } + if c.ConnectRetryTimeout.Duration != 0 { + connectRetryTimeout = &c.ConnectRetryTimeout + } if c.TLSTimeout != defaultTLSTimeout { tlsTimeout = &c.TLSTimeout } @@ -522,6 +544,7 @@ func ConvertToRawOriginConfig(c OriginRequestConfig) config.OriginRequestConfig return config.OriginRequestConfig{ ConnectTimeout: connectTimeout, + ConnectRetryTimeout: connectRetryTimeout, TLSTimeout: tlsTimeout, TCPKeepAlive: tcpKeepAlive, NoHappyEyeballs: defaultBoolToNil(c.NoHappyEyeballs), @@ -544,7 +567,7 @@ func ConvertToRawOriginConfig(c OriginRequestConfig) config.OriginRequestConfig } func convertToRawIPRules(ipRules []ipaccess.Rule) []config.IngressIPRule { - result := make([]config.IngressIPRule, 0) + result := make([]config.IngressIPRule, 0, len(ipRules)) for _, r := range ipRules { cidr := r.StringCIDR() diff --git a/ingress/ingress.go b/ingress/ingress.go index a325271a7e7..62df7b7272d 100644 --- a/ingress/ingress.go +++ b/ingress/ingress.go @@ -19,12 +19,13 @@ import ( ) var ( - ErrNoIngressRules = errors.New("The config file doesn't contain any ingress rules") - ErrNoIngressRulesCLI = errors.New("No ingress rules were defined in provided config (if any) nor from the cli, cloudflared will return 503 for all incoming HTTP requests") - errLastRuleNotCatchAll = errors.New("The last ingress rule must match all URLs (i.e. it should not have a hostname or path filter)") - errBadWildcard = errors.New("Hostname patterns can have at most one wildcard character (\"*\") and it can only be used for subdomains, e.g. \"*.example.com\"") - errHostnameContainsPort = errors.New("Hostname cannot contain a port") - ErrURLIncompatibleWithIngress = errors.New("You can't set the --url flag (or $TUNNEL_URL) when using multiple-origin ingress rules") + ErrNoIngressRules = errors.New("The config file doesn't contain any ingress rules") + ErrNoIngressRulesCLI = errors.New("No ingress rules were defined in provided config (if any) nor from the cli, cloudflared will return 503 for all incoming HTTP requests") + errLastRuleNotCatchAll = errors.New("The last ingress rule must match all URLs (i.e. it should not have a hostname or path filter)") + errBadWildcard = errors.New("Hostname patterns can have at most one wildcard character (\"*\") and it can only be used for subdomains, e.g. \"*.example.com\"") + errHostnameContainsPort = errors.New("Hostname cannot contain a port") + errNegativeConnectRetryTimeout = errors.New("connectRetryTimeout must not be negative") + ErrURLIncompatibleWithIngress = errors.New("You can't set the --url flag (or $TUNNEL_URL) when using multiple-origin ingress rules") ) const ( @@ -135,6 +136,9 @@ func parseCLIIngress(c *cli.Context, allowURLFromArgs bool) (Ingress, error) { // Construct an Ingress with the single rule. defaults := originRequestFromSingleRule(c) + if defaults.ConnectRetryTimeout.Duration < 0 { + return Ingress{}, errNegativeConnectRetryTimeout + } ing := Ingress{ Rules: []Rule{ { @@ -243,9 +247,15 @@ func validateAccessConfiguration(cfg *config.AccessConfig) error { } func validateIngress(ingress []config.UnvalidatedIngressRule, defaults OriginRequestConfig) (Ingress, error) { + if defaults.ConnectRetryTimeout.Duration < 0 { + return Ingress{}, errNegativeConnectRetryTimeout + } rules := make([]Rule, len(ingress)) for i, r := range ingress { cfg := setConfig(defaults, r.OriginRequest) + if cfg.ConnectRetryTimeout.Duration < 0 { + return Ingress{}, fmt.Errorf("ingress rule %d: %w", i+1, errNegativeConnectRetryTimeout) + } var service OriginService if prefix := "unix:"; strings.HasPrefix(r.Service, prefix) { diff --git a/ingress/origin_connection_test.go b/ingress/origin_connection_test.go index b031c011301..78960c10c7f 100644 --- a/ingress/origin_connection_test.go +++ b/ingress/origin_connection_test.go @@ -165,7 +165,7 @@ func TestSocksStreamWSOverTCPConnection(t *testing.T) { require.NoError(t, err) transport := &http.Transport{ - Dial: eyeballDialer.Dial, + DialContext: eyeballDialer.(proxy.ContextDialer).DialContext, } // Request URL doesn't matter because the transport is using eyeballDialer to connectq diff --git a/ingress/origin_retry.go b/ingress/origin_retry.go new file mode 100644 index 00000000000..7abb53712c9 --- /dev/null +++ b/ingress/origin_retry.go @@ -0,0 +1,74 @@ +package ingress + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "syscall" + "time" + + "github.com/cloudflare/cloudflared/retry" +) + +// HTTP transport with request-scoped cancellation for optional connection retries. +type originHTTPTransport struct { + *http.Transport + connectRetryTimeout time.Duration +} + +type originDialContextKey struct{} + +func (t *originHTTPTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if t.connectRetryTimeout <= 0 { + return t.Transport.RoundTrip(req) + } + + // Preserving cancellation across net/http's detached dial context; a request + // that finishes using a pooled connection must also stop its pending retries. + dialCtx, cancel := context.WithCancel(req.Context()) + defer cancel() + ctx := context.WithValue(req.Context(), originDialContextKey{}, dialCtx) + return t.Transport.RoundTrip(req.WithContext(ctx)) +} + +// Dial the origin within a bounded retry window, before any request is written. +// Only absent listeners are retried; TLS and HTTP errors remain the transport's responsibility. +func dialOriginWithRetry( + ctx context.Context, + network, address string, + timeout time.Duration, + dial func(context.Context, string, string) (net.Conn, error), +) (net.Conn, error) { + if timeout <= 0 { + return dial(ctx, network, address) + } + + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + if requestCtx, ok := ctx.Value(originDialContextKey{}).(context.Context); ok { + stop := context.AfterFunc(requestCtx, cancel) + defer stop() + err := requestCtx.Err() + if err != nil { + return nil, err + } + } + + // Capping jittered exponential backoff at 80ms to cover short local restarts. + backoff := retry.NewBackoff(3, 10*time.Millisecond, true) + for { + conn, err := dial(ctx, network, address) + if err == nil { + return conn, nil + } + if !errors.Is(err, errOriginConnectionRefused) && (network != "unix" || !errors.Is(err, syscall.ENOENT)) { + return nil, err + } + again := backoff.Backoff(ctx) + if !again { + return nil, fmt.Errorf("origin connection retry stopped: %w (last dial error: %w)", ctx.Err(), err) + } + } +} diff --git a/ingress/origin_retry_config_test.go b/ingress/origin_retry_config_test.go new file mode 100644 index 00000000000..e6226286d2a --- /dev/null +++ b/ingress/origin_retry_config_test.go @@ -0,0 +1,99 @@ +package ingress + +import ( + "encoding/json" + "testing" + "time" + + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v3" + + "github.com/cloudflare/cloudflared/config" +) + +func TestOriginConnectRetryConfig(t *testing.T) { + t.Parallel() + for _, format := range []string{"yaml", "json"} { + t.Run(format, func(t *testing.T) { + t.Parallel() + var ing Ingress + if format == "yaml" { + var cfg config.Configuration + err := yaml.Unmarshal([]byte(` +originRequest: + connectRetryTimeout: 2s +ingress: + - hostname: inherited.example.com + service: unix:/tmp/app.sock + - hostname: overridden.example.com + service: http://localhost:8000 + originRequest: + connectRetryTimeout: 500ms + - service: unix+tls:/tmp/app.sock + originRequest: + connectRetryTimeout: 0s +`), &cfg) + require.NoError(t, err) + ing, err = ParseIngress(&cfg) + require.NoError(t, err) + } else { + var cfg RemoteConfig + err := json.Unmarshal([]byte(`{"originRequest":{"connectRetryTimeout":2},"ingress":[ +{"hostname":"inherited.example.com","service":"unix:/tmp/app.sock"}, +{"hostname":"overridden.example.com","service":"http://localhost:8000","originRequest":{"connectRetryTimeout":0.5}}, +{"service":"unix+tls:/tmp/app.sock","originRequest":{"connectRetryTimeout":0}}]}`), &cfg) + require.NoError(t, err) + ing = cfg.Ingress + } + require.Equal(t, 2*time.Second, ing.Defaults.ConnectRetryTimeout.Duration) + require.Len(t, ing.Rules, 3) + for i, timeout := range []time.Duration{2 * time.Second, 500 * time.Millisecond, 0} { + require.Equal(t, timeout, ing.Rules[i].Config.ConnectRetryTimeout.Duration) + } + }) + } +} + +func TestOriginConnectRetryInvalidConfig(t *testing.T) { + t.Parallel() + testCases := []struct { + name string + yaml string + json string + }{ + { + name: "negative global timeout", + yaml: "originRequest: {connectRetryTimeout: -1s}\ningress: [{service: 'unix:/tmp/app.sock'}]", + json: `{"originRequest":{"connectRetryTimeout":-1},"ingress":[{"service":"unix:/tmp/app.sock"}]}`, + }, + { + name: "negative rule timeout", + yaml: "ingress: [{service: 'unix:/tmp/app.sock', originRequest: {connectRetryTimeout: -1s}}]", + json: `{"ingress":[{"service":"unix:/tmp/app.sock","originRequest":{"connectRetryTimeout":-1}}]}`, + }, + { + name: "override does not hide invalid global timeout", + yaml: "originRequest: {connectRetryTimeout: -1s}\ningress: [{service: 'unix:/tmp/app.sock', originRequest: {connectRetryTimeout: 0s}}]", + json: `{"originRequest":{"connectRetryTimeout":-1},"ingress":[{"service":"unix:/tmp/app.sock","originRequest":{"connectRetryTimeout":0}}]}`, + }, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + t.Run("yaml", func(t *testing.T) { + t.Parallel() + var cfg config.Configuration + err := yaml.Unmarshal([]byte(tc.yaml), &cfg) + require.NoError(t, err) + _, err = ParseIngress(&cfg) + require.ErrorContains(t, err, "connectRetryTimeout must not be negative") + }) + t.Run("json", func(t *testing.T) { + t.Parallel() + var cfg RemoteConfig + err := json.Unmarshal([]byte(tc.json), &cfg) + require.ErrorContains(t, err, "connectRetryTimeout must not be negative") + }) + }) + } +} diff --git a/ingress/origin_retry_http_test.go b/ingress/origin_retry_http_test.go new file mode 100644 index 00000000000..e532589e80a --- /dev/null +++ b/ingress/origin_retry_http_test.go @@ -0,0 +1,221 @@ +package ingress + +import ( + "context" + "crypto/tls" + "io" + "net" + "net/http" + "net/http/httptest" + "net/http/httptrace" + "net/url" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/cloudflare/cloudflared/config" +) + +func TestHTTPOriginWaitsForListener(t *testing.T) { + t.Parallel() + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + address := listener.Addr().String() + err = listener.Close() + require.NoError(t, err) + originURL, err := url.Parse("http://" + address) + require.NoError(t, err) + origin := &httpService{url: originURL} + cfg := originRequestFromConfig(config.OriginRequestConfig{}) + cfg.ConnectRetryTimeout.Duration = 2 * time.Second + err = origin.start(TestLogger, t.Context().Done(), cfg) + require.NoError(t, err) + t.Cleanup(origin.transport.CloseIdleConnections) + failed := make(chan struct{}) + var once sync.Once + trace := &httptrace.ClientTrace{ConnectDone: func(_, _ string, err error) { + if err != nil { + once.Do(func() { close(failed) }) + } + }} + req, err := http.NewRequestWithContext(httptrace.WithClientTrace(t.Context(), trace), http.MethodGet, "http://localhost/", nil) + require.NoError(t, err) + results := make(chan error, 1) + go func() { + resp, err := origin.RoundTrip(req) + if err != nil { + results <- err + return + } + defer func() { _ = resp.Body.Close() }() + assert.Equal(t, http.StatusOK, resp.StatusCode) + body, err := io.ReadAll(resp.Body) + assert.Equal(t, "origin ready", string(body)) + results <- err + }() + select { + case <-failed: + case <-time.After(5 * time.Second): + t.Fatal("no refused connection") + } + listener, err = net.Listen("tcp", address) + require.NoError(t, err) + server := &httptest.Server{Listener: listener, Config: &http.Server{ReadHeaderTimeout: time.Second, Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, err := io.WriteString(w, "origin ready") + assert.NoError(t, err) + })}} + server.Start() + defer server.Close() + select { + case err := <-results: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("origin connection did not recover") + } +} + +func TestHTTPOriginRetryDoesNotRetryTLSOrHTTPFailures(t *testing.T) { + t.Parallel() + for _, tlsFailure := range []bool{false, true} { + name := "HTTP 503" + if tlsFailure { + name = "invalid certificate" + } + t.Run(name, func(t *testing.T) { + t.Parallel() + var received atomic.Int32 + server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + received.Add(1) + w.WriteHeader(http.StatusServiceUnavailable) + _, err := io.WriteString(w, "temporarily unavailable") + assert.NoError(t, err) + })) + if tlsFailure { + server.StartTLS() + } else { + server.Start() + } + defer server.Close() + originURL, err := url.Parse(server.URL) + require.NoError(t, err) + origin := &httpService{url: originURL} + cfg := originRequestFromConfig(config.OriginRequestConfig{}) + cfg.ConnectRetryTimeout.Duration = time.Second + err = origin.start(TestLogger, t.Context().Done(), cfg) + require.NoError(t, err) + defer origin.transport.CloseIdleConnections() + var attempts atomic.Int32 + trace := &httptrace.ClientTrace{ConnectStart: func(_, _ string) { attempts.Add(1) }} + req, err := http.NewRequestWithContext(httptrace.WithClientTrace(t.Context(), trace), http.MethodGet, "http://localhost/", nil) + require.NoError(t, err) + resp, err := origin.RoundTrip(req) + if tlsFailure { + var certErr *tls.CertificateVerificationError + require.ErrorAs(t, err, &certErr) + require.Nil(t, resp) + require.Zero(t, received.Load()) + } else { + require.NoError(t, err) + defer func() { _ = resp.Body.Close() }() + require.Equal(t, http.StatusServiceUnavailable, resp.StatusCode) + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Equal(t, "temporarily unavailable", string(body)) + require.EqualValues(t, 1, received.Load()) + } + require.EqualValues(t, 1, attempts.Load()) + }) + } +} + +func TestHTTPOriginRetryStopsUnusedDial(t *testing.T) { + t.Parallel() + firstArrived := make(chan struct{}) + releaseFirst := make(chan struct{}) + var release sync.Once + defer release.Do(func() { close(releaseFirst) }) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/first" { + close(firstArrived) + <-releaseFirst + } + _, err := io.WriteString(w, "ok") + assert.NoError(t, err) + })) + t.Cleanup(server.Close) + originURL, err := url.Parse(server.URL) + require.NoError(t, err) + origin := &httpService{url: originURL} + cfg := originRequestFromConfig(config.OriginRequestConfig{}) + cfg.ConnectRetryTimeout.Duration = time.Minute + err = origin.start(TestLogger, t.Context().Done(), cfg) + require.NoError(t, err) + defer origin.transport.CloseIdleConnections() + dial := origin.transport.DialContext + stopped := make(chan struct{}) + origin.transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { + conn, err := dial(ctx, network, address) + if err != nil { + close(stopped) + } + return conn, err + } + failed := make(chan struct{}) + var once sync.Once + trace := &httptrace.ClientTrace{ConnectDone: func(_, _ string, err error) { + if err != nil { + once.Do(func() { close(failed) }) + } + }} + results := make(chan error, 2) + send := func(path string) { + req, err := http.NewRequestWithContext(httptrace.WithClientTrace(t.Context(), trace), http.MethodGet, "http://localhost"+path, nil) + if err != nil { + results <- err + return + } + resp, err := origin.RoundTrip(req) + if err != nil { + results <- err + return + } + defer func() { _ = resp.Body.Close() }() + assert.Equal(t, http.StatusOK, resp.StatusCode) + body, err := io.ReadAll(resp.Body) + assert.Equal(t, "ok", string(body)) + results <- err + } + go send("/first") + select { + case <-firstArrived: + case <-time.After(5 * time.Second): + t.Fatal("first request did not arrive") + } + // Closing only the listener keeps the first connection available for reuse. + err = server.Listener.Close() + require.NoError(t, err) + go send("/second") + select { + case <-failed: + case <-time.After(5 * time.Second): + t.Fatal("second request did not encounter the absent listener") + } + release.Do(func() { close(releaseFirst) }) + for range 2 { + select { + case err := <-results: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("request did not finish") + } + } + select { + case <-stopped: + case <-time.After(time.Second): + t.Fatal("unused dial outlived its request") + } +} diff --git a/ingress/origin_retry_posix.go b/ingress/origin_retry_posix.go new file mode 100644 index 00000000000..50a5f3f889b --- /dev/null +++ b/ingress/origin_retry_posix.go @@ -0,0 +1,7 @@ +//go:build !windows + +package ingress + +import "syscall" + +const errOriginConnectionRefused = syscall.ECONNREFUSED diff --git a/ingress/origin_retry_posix_test.go b/ingress/origin_retry_posix_test.go new file mode 100644 index 00000000000..59271715890 --- /dev/null +++ b/ingress/origin_retry_posix_test.go @@ -0,0 +1,351 @@ +//go:build !windows + +package ingress + +import ( + "context" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/http/httptrace" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/cloudflare/cloudflared/config" +) + +func newRetryUnixOrigin(t *testing.T, timeout time.Duration, scheme string) *unixSocketPath { + t.Helper() + // Keeping the socket pathname below the Unix sockaddr limit on macOS as well as Linux. + dir, err := os.MkdirTemp("", "cfd-retry-") //nolint:usetesting // t.TempDir includes the test name and can exceed sockaddr_un limits. + require.NoError(t, err) + t.Cleanup(func() { err := os.RemoveAll(dir); assert.NoError(t, err) }) + origin := &unixSocketPath{path: filepath.Join(dir, "http.sock"), scheme: scheme} + cfg := originRequestFromConfig(config.OriginRequestConfig{}) + cfg.ConnectRetryTimeout.Duration = timeout + cfg.NoTLSVerify = true + err = origin.start(TestLogger, t.Context().Done(), cfg) + require.NoError(t, err) + t.Cleanup(origin.transport.CloseIdleConnections) + return origin +} + +func startRetryUnixServer(t *testing.T, origin *unixSocketPath, handler http.Handler, http2 bool) *httptest.Server { + t.Helper() + listener, err := net.Listen("unix", origin.path) + require.NoError(t, err) + server := &httptest.Server{Listener: listener, Config: &http.Server{Handler: handler, ReadHeaderTimeout: time.Second}, EnableHTTP2: http2} + if origin.scheme == "https" { + server.StartTLS() + } else { + server.Start() + } + t.Cleanup(server.Close) + return server +} + +func TestUnixOriginWaitsForListener(t *testing.T) { + t.Parallel() + for _, mode := range []string{"http", "https", "http2", "stale socket"} { + t.Run(mode, func(t *testing.T) { + t.Parallel() + scheme := "http" + if mode == "https" || mode == "http2" { + scheme = "https" + } + origin := newRetryUnixOrigin(t, 2*time.Second, scheme) + origin.transport.ForceAttemptHTTP2 = mode == "http2" + if mode == "stale socket" { + listener, err := net.ListenUnix("unix", &net.UnixAddr{Name: origin.path, Net: "unix"}) + require.NoError(t, err) + listener.SetUnlinkOnClose(false) + err = listener.Close() + require.NoError(t, err) + } + const arrivals = 8 + failedDial := make(chan struct{}, arrivals) + results := make(chan error, arrivals) + var reads atomic.Int32 + var received atomic.Int32 + for i := range arrivals { + go func() { + var once sync.Once + trace := &httptrace.ClientTrace{ConnectDone: func(_, _ string, err error) { + if err != nil { + once.Do(func() { failedDial <- struct{}{} }) + } + }} + ctx := httptrace.WithClientTrace(t.Context(), trace) + // A streaming, non-replayable POST body: GetBody remains nil. + body := &countingRetryBody{Reader: strings.NewReader(fmt.Sprintf("payload-%d", i)), reads: &reads} + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://localhost/", body) + if err != nil { + results <- err + return + } + resp, err := origin.RoundTrip(req) + if err != nil { + results <- err + return + } + defer func() { _ = resp.Body.Close() }() + data, err := io.ReadAll(resp.Body) + if err == nil && string(data) != fmt.Sprintf("payload-%d", i) { + err = fmt.Errorf("incorrect body: %q", data) + } + results <- err + }() + } + for range arrivals { + select { + case <-failedDial: + case <-time.After(5 * time.Second): + t.Fatal("request never attempted its connection") + } + } + require.Zero(t, reads.Load(), "request bodies must remain unread while the listener is absent") + if mode == "stale socket" { + err := os.Remove(origin.path) + require.NoError(t, err) + } + startRetryUnixServer(t, origin, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + received.Add(1) + assert.Equal(t, http.MethodPost, r.Method) + if mode == "http2" { + assert.Equal(t, 2, r.ProtoMajor) + } + _, err := io.Copy(w, r.Body) + assert.NoError(t, err) + }), mode == "http2") + for range arrivals { + select { + case err := <-results: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("request did not resume after the listener appeared") + } + } + require.EqualValues(t, arrivals, received.Load(), "each POST must arrive exactly once") + }) + } +} + +type countingRetryBody struct { + io.Reader + reads *atomic.Int32 +} + +func (b *countingRetryBody) Read(p []byte) (int, error) { + b.reads.Add(1) + return b.Reader.Read(p) +} + +func TestUnixOriginRetryDisabledAndExpired(t *testing.T) { + t.Parallel() + for _, timeout := range []time.Duration{0, 50 * time.Millisecond} { + name := "disabled" + if timeout > 0 { + name = "retry window exhausted" + } + t.Run(name, func(t *testing.T) { + t.Parallel() + origin := newRetryUnixOrigin(t, timeout, "http") + var attempts atomic.Int32 + trace := &httptrace.ClientTrace{ConnectStart: func(_, _ string) { attempts.Add(1) }} + ctx := httptrace.WithClientTrace(t.Context(), trace) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost/", nil) + require.NoError(t, err) + resp, err := origin.RoundTrip(req) + if resp != nil { + defer func() { _ = resp.Body.Close() }() + } + require.Error(t, err) + require.Nil(t, resp) + if timeout == 0 { + require.ErrorIs(t, err, os.ErrNotExist) + require.NotErrorIs(t, err, context.DeadlineExceeded) + require.EqualValues(t, 1, attempts.Load()) + return + } + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Greater(t, attempts.Load(), int32(1)) + }) + } +} + +func TestUnixOriginRetryStopsWhenRequestEnds(t *testing.T) { + t.Parallel() + testCases := []struct { + name string + timeout time.Duration + err error + }{ + {"request canceled", time.Minute, context.Canceled}, + {"request deadline exceeded", 250 * time.Millisecond, context.DeadlineExceeded}, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + origin := newRetryUnixOrigin(t, time.Minute, "http") + dial := origin.transport.DialContext + dialStopped := make(chan struct{}) + origin.transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { + defer close(dialStopped) + return dial(ctx, network, address) + } + failed := make(chan struct{}) + var once sync.Once + trace := &httptrace.ClientTrace{ConnectDone: func(_, _ string, err error) { + if err != nil { + once.Do(func() { close(failed) }) + } + }} + ctx, cancel := context.WithTimeout(t.Context(), tc.timeout) + defer cancel() + req, err := http.NewRequestWithContext(httptrace.WithClientTrace(ctx, trace), http.MethodGet, "http://localhost/", nil) + require.NoError(t, err) + result := make(chan error, 1) + go func() { + resp, err := origin.RoundTrip(req) + if resp != nil { + _ = resp.Body.Close() + } + result <- err + }() + select { + case <-failed: + case <-time.After(5 * time.Second): + t.Fatal("request did not encounter the absent listener") + } + if tc.err == context.Canceled { + cancel() + } + select { + case err := <-result: + require.ErrorIs(t, err, tc.err) + case <-time.After(time.Second): + t.Fatal("request ignored cancellation") + } + select { + case <-dialStopped: + case <-time.After(time.Second): + t.Fatal("dial kept retrying after request ended") + } + }) + } +} + +func TestUnixOriginRetriesAfterPooledConnectionCloses(t *testing.T) { + t.Parallel() + origin := newRetryUnixOrigin(t, 2*time.Second, "http") + old := startRetryUnixServer(t, origin, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, err := io.WriteString(w, "old") + assert.NoError(t, err) + }), false) + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://localhost/", nil) + require.NoError(t, err) + resp, err := origin.RoundTrip(req) + require.NoError(t, err) + _, err = io.Copy(io.Discard, resp.Body) + require.NoError(t, err) + err = resp.Body.Close() + require.NoError(t, err) + old.Close() + failed := make(chan struct{}) + var once sync.Once + trace := &httptrace.ClientTrace{ConnectDone: func(_, _ string, err error) { + if err != nil { + once.Do(func() { close(failed) }) + } + }} + req, err = http.NewRequestWithContext(httptrace.WithClientTrace(t.Context(), trace), http.MethodGet, "http://localhost/", nil) + require.NoError(t, err) + result := make(chan error, 1) + go func() { + resp, err := origin.RoundTrip(req) + if err != nil { + result <- err + return + } + defer func() { _ = resp.Body.Close() }() + data, err := io.ReadAll(resp.Body) + if err == nil && string(data) != "new" { + err = fmt.Errorf("incorrect release: %q", data) + } + result <- err + }() + select { + case <-failed: + case <-time.After(5 * time.Second): + t.Fatal("no connection attempted during restart") + } + startRetryUnixServer(t, origin, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, err := io.WriteString(w, "new") + assert.NoError(t, err) + }), false) + select { + case err := <-result: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("request did not reach new instance") + } +} + +func TestUnixOriginDoesNotReplayDeliveredPOST(t *testing.T) { + t.Parallel() + origin := newRetryUnixOrigin(t, time.Second, "http") + var delivered atomic.Int32 + startRetryUnixServer(t, origin, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, err := io.Copy(io.Discard, r.Body) + assert.NoError(t, err) + delivered.Add(1) + conn, _, err := w.(http.Hijacker).Hijack() + if !assert.NoError(t, err) { + return + } + err = conn.Close() + assert.NoError(t, err) + }), false) + req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, "http://localhost/", strings.NewReader("operation")) + require.NoError(t, err) + resp, err := origin.RoundTrip(req) + if resp != nil { + defer func() { _ = resp.Body.Close() }() + } + require.Error(t, err) + require.Nil(t, resp) + require.EqualValues(t, 1, delivered.Load()) +} + +func TestUnixOriginRetryDoesNotCancelResponseBody(t *testing.T) { + t.Parallel() + origin := newRetryUnixOrigin(t, time.Second, "http") + release := make(chan struct{}) + defer close(release) + startRetryUnixServer(t, origin, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + <-release + _, err := io.WriteString(w, "streamed response") + assert.NoError(t, err) + }), false) + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://localhost/", nil) + require.NoError(t, err) + resp, err := origin.RoundTrip(req) + require.NoError(t, err) + defer func() { _ = resp.Body.Close() }() + release <- struct{}{} + data, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Equal(t, "streamed response", string(data)) +} diff --git a/ingress/origin_retry_test.go b/ingress/origin_retry_test.go new file mode 100644 index 00000000000..e99f2fda5d6 --- /dev/null +++ b/ingress/origin_retry_test.go @@ -0,0 +1,93 @@ +package ingress + +import ( + "context" + "net" + "syscall" + "testing" + "testing/synctest" + "time" + + "github.com/stretchr/testify/require" +) + +func TestDialOriginRetryReturnsPermanentErrors(t *testing.T) { + t.Parallel() + // Injecting OS errors avoids platform-specific permissions and DNS dependencies. + testCases := []struct { + name string + network string + err error + initiallyRefused bool + }{ + {"permission denied", "unix", syscall.EACCES, false}, + {"not a directory", "unix", syscall.ENOTDIR, false}, + {"invalid address", "unix", syscall.EINVAL, false}, + {"network unreachable", "tcp", syscall.ENETUNREACH, false}, + {"DNS failure", "tcp", &net.DNSError{Err: "no such host", Name: "origin", IsNotFound: true}, false}, + {"dial timeout", "tcp", context.DeadlineExceeded, false}, + {"permanent failure after refused connection", "unix", syscall.EACCES, true}, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + attempts := 0 + dial := func(context.Context, string, string) (net.Conn, error) { + attempts++ + if tc.initiallyRefused && attempts == 1 { + return nil, &net.OpError{Op: "dial", Net: tc.network, Err: errOriginConnectionRefused} + } + return nil, &net.OpError{Op: "dial", Net: tc.network, Err: tc.err} + } + conn, err := dialOriginWithRetry(t.Context(), tc.network, "origin", time.Second, dial) + require.Nil(t, conn) + require.ErrorIs(t, err, tc.err) + expectedAttempts := 1 + if tc.initiallyRefused { + expectedAttempts++ + } + require.Equal(t, expectedAttempts, attempts, "permanent failures must not trigger another dial") + }) + }) + } +} + +func TestDialOriginRetryBoundsConnectionTime(t *testing.T) { + t.Parallel() + testCases := []struct { + name string + retryTimeout time.Duration + requestTimeout time.Duration + firstRefusal time.Duration + }{ + {"retry timeout during first dial", 50 * time.Millisecond, time.Second, 0}, + {"retry timeout includes previous dials", 50 * time.Millisecond, time.Second, 25 * time.Millisecond}, + {"shorter request deadline", time.Second, 50 * time.Millisecond, 0}, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + refused := false + dial := func(ctx context.Context, _, _ string) (net.Conn, error) { + if tc.firstRefusal > 0 && !refused { + refused = true + time.Sleep(tc.firstRefusal) + return nil, errOriginConnectionRefused + } + <-ctx.Done() + return nil, ctx.Err() + } + ctx, cancel := context.WithTimeout(t.Context(), tc.requestTimeout) + defer cancel() + start := time.Now() + conn, err := dialOriginWithRetry(ctx, "tcp", "origin", tc.retryTimeout, dial) + elapsed := time.Since(start) + require.Nil(t, conn) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Equal(t, 50*time.Millisecond, elapsed) + }) + }) + } +} diff --git a/ingress/origin_retry_windows.go b/ingress/origin_retry_windows.go new file mode 100644 index 00000000000..19b0e3ff58a --- /dev/null +++ b/ingress/origin_retry_windows.go @@ -0,0 +1,5 @@ +package ingress + +import "golang.org/x/sys/windows" + +const errOriginConnectionRefused = windows.WSAECONNREFUSED diff --git a/ingress/origin_service.go b/ingress/origin_service.go index e13204c5789..5e33ddeea97 100644 --- a/ingress/origin_service.go +++ b/ingress/origin_service.go @@ -43,7 +43,7 @@ type OriginService interface { type unixSocketPath struct { path string scheme string - transport *http.Transport + transport *originHTTPTransport } func (o *unixSocketPath) String() string { @@ -70,7 +70,7 @@ func (o unixSocketPath) MarshalJSON() ([]byte, error) { type httpService struct { url *url.URL hostHeader string - transport *http.Transport + transport *originHTTPTransport matchSNIToHost bool } @@ -99,7 +99,6 @@ type rawTCPService struct { name string dialer net.Dialer writeTimeout time.Duration - logger *zerolog.Logger } func (o *rawTCPService) String() string { @@ -233,10 +232,10 @@ func (o *helloWorld) start( if err != nil { return errors.Wrap(err, "Cannot start Hello World Server") } - go hello.StartHelloWorldServer(log, helloListener, shutdownC) + go hello.StartHelloWorldServer(log, helloListener, shutdownC) //nolint:errcheck // The managed server exits asynchronously when shutdownC closes. o.server = helloListener - o.httpService.url = &url.URL{ + o.url = &url.URL{ Scheme: "https", Host: o.server.Addr().String(), } @@ -343,21 +342,24 @@ func (nrc *NopReadCloser) Close() error { return nil } -func newHTTPTransport(service OriginService, cfg OriginRequestConfig, log *zerolog.Logger) (*http.Transport, error) { +func newHTTPTransport(service OriginService, cfg OriginRequestConfig, log *zerolog.Logger) (*originHTTPTransport, error) { originCertPool, err := tlsconfig.LoadOriginCA(cfg.CAPool, log) if err != nil { return nil, errors.Wrap(err, "Error loading cert pool") } - httpTransport := http.Transport{ - Proxy: http.ProxyFromEnvironment, - MaxIdleConns: cfg.KeepAliveConnections, - MaxIdleConnsPerHost: cfg.KeepAliveConnections, - IdleConnTimeout: cfg.KeepAliveTimeout.Duration, - TLSHandshakeTimeout: cfg.TLSTimeout.Duration, - ExpectContinueTimeout: 1 * time.Second, - TLSClientConfig: &tls.Config{RootCAs: originCertPool, InsecureSkipVerify: cfg.NoTLSVerify}, - ForceAttemptHTTP2: cfg.Http2Origin, + httpTransport := originHTTPTransport{ + Transport: &http.Transport{ + Proxy: http.ProxyFromEnvironment, + MaxIdleConns: cfg.KeepAliveConnections, + MaxIdleConnsPerHost: cfg.KeepAliveConnections, + IdleConnTimeout: cfg.KeepAliveTimeout.Duration, + TLSHandshakeTimeout: cfg.TLSTimeout.Duration, + ExpectContinueTimeout: 1 * time.Second, + TLSClientConfig: &tls.Config{RootCAs: originCertPool, InsecureSkipVerify: cfg.NoTLSVerify}, //nolint:gosec // Explicit origin noTLSVerify configuration. + ForceAttemptHTTP2: cfg.Http2Origin, + }, + connectRetryTimeout: cfg.ConnectRetryTimeout.Duration, } if _, isHelloWorld := service.(*helloWorld); !isHelloWorld && cfg.OriginServerName != "" { httpTransport.TLSClientConfig.ServerName = cfg.OriginServerName @@ -372,9 +374,10 @@ func newHTTPTransport(service OriginService, cfg OriginRequestConfig, log *zerol } // DialContext depends on which kind of origin is being used. - dialContext := dialer.DialContext + dialContext := func(ctx context.Context, network, address string) (net.Conn, error) { + return dialOriginWithRetry(ctx, network, address, httpTransport.connectRetryTimeout, dialer.DialContext) + } switch service := service.(type) { - // If this origin is a unix socket, enforce network type "unix". case *unixSocketPath: httpTransport.DialContext = func(ctx context.Context, _, _ string) (net.Conn, error) { diff --git a/orchestration/config_test.go b/orchestration/config_test.go index affac27a978..1f00ee2fbb5 100644 --- a/orchestration/config_test.go +++ b/orchestration/config_test.go @@ -14,10 +14,12 @@ import ( // TestNewLocalConfig_MarshalJSON tests that we are able to converte a compiled and validated config back // into an "unvalidated" format which is compatible with Remote Managed configurations. func TestNewLocalConfig_MarshalJSON(t *testing.T) { + t.Parallel() rawConfig := []byte(` { "originRequest": { "connectTimeout": 160, + "connectRetryTimeout": 2, "httpHostHeader": "default" }, "ingress": [ @@ -26,10 +28,11 @@ func TestNewLocalConfig_MarshalJSON(t *testing.T) { "service": "https://localhost:8000" }, { - "hostname": "*", + "hostname": "overridden.example.com", "service": "https://localhost:8001", "originRequest": { "connectTimeout": 121, + "connectRetryTimeout": 0.5, "tlsTimeout": 2, "noHappyEyeballs": false, "tcpKeepAlive": 2, @@ -57,6 +60,13 @@ func TestNewLocalConfig_MarshalJSON(t *testing.T) { } ] } + }, + { + "hostname": "*", + "service": "unix:/tmp/app.sock", + "originRequest": { + "connectRetryTimeout": 0 + } } ], "warp-routing": { @@ -68,6 +78,10 @@ func TestNewLocalConfig_MarshalJSON(t *testing.T) { var expectedConfig ingress.RemoteConfig err := json.Unmarshal(rawConfig, &expectedConfig) require.NoError(t, err) + require.Len(t, expectedConfig.Ingress.Rules, 3) + for i, timeout := range []time.Duration{2 * time.Second, 500 * time.Millisecond, 0} { + require.Equal(t, timeout, expectedConfig.Ingress.Rules[i].Config.ConnectRetryTimeout.Duration) + } c := &newLocalConfig{ RemoteConfig: expectedConfig, @@ -81,13 +95,13 @@ func TestNewLocalConfig_MarshalJSON(t *testing.T) { err = json.Unmarshal(jsonSerde, &remoteConfig) require.NoError(t, err) - require.Equal(t, remoteConfig.WarpRouting, ingress.WarpRoutingConfig{ + require.Equal(t, ingress.WarpRoutingConfig{ ConnectTimeout: config.CustomDuration{ Duration: time.Second, }, TCPKeepAlive: config.CustomDuration{ Duration: 30 * time.Second, // default value is 30 seconds }, - }) - require.Equal(t, remoteConfig.Ingress.Rules, expectedConfig.Ingress.Rules) + }, remoteConfig.WarpRouting) + require.Equal(t, expectedConfig.Ingress.Rules, remoteConfig.Ingress.Rules) }