add http transport options to http processor, http_client input & output

jem-davies · jem-davies · commit dae3a18b7ab8 · 2025-11-21T18:03:24.000Z
Signed-off-by: Jem Davies &lt;jemsot@gmail.com&gt;
diff --git a/internal/httpclient/client.go b/internal/httpclient/client.go
@@ -57,9 +57,20 @@ func NewClientFromOldConfig(conf OldConfig, mgr *service.Resources, opts ...Requ
 		return nil, err
 	}
 
+	var transport *http.Transport
+	if conf.customTransport {
+		transport = conf.transport
+	} else {
+		if t, ok := http.DefaultTransport.(*http.Transport); ok {
+			transport = t.Clone()
+		} else {
+			return nil, fmt.Errorf("unable to use http.DefaultTransport, unexpected type: %T", http.DefaultTransport)
+		}
+	}
+
 	h := Client{
 		reqCreator:        reqCreator,
-		client:            &http.Client{},
+		client:            &http.Client{Transport: transport},
 		metaExtractFilter: conf.ExtractMetadata,
 
 		backoffOn: map[int]struct{}{},
@@ -76,7 +87,7 @@ func NewClientFromOldConfig(conf OldConfig, mgr *service.Resources, opts ...Requ
 	}
 
 	if conf.TLSEnabled && conf.TLSConf != nil {
-		if c, ok := http.DefaultTransport.(*http.Transport); ok {
+		if c, ok := h.client.Transport.(*http.Transport); ok {
 			cloned := c.Clone()
 			cloned.TLSClientConfig = conf.TLSConf
 			h.client.Transport = cloned
@@ -94,7 +105,9 @@ func NewClientFromOldConfig(conf OldConfig, mgr *service.Resources, opts ...Requ
 		}
 		if h.client.Transport != nil {
 			if tr, ok := h.client.Transport.(*http.Transport); ok {
-				tr.Proxy = http.ProxyURL(proxyURL)
+				cloned := tr.Clone()
+				cloned.Proxy = http.ProxyURL(proxyURL)
+				h.client.Transport = cloned
 			} else {
 				return nil, fmt.Errorf("unable to apply proxy_url to transport, unexpected type %T", h.client.Transport)
 			}
diff --git a/internal/httpclient/config.go b/internal/httpclient/config.go
@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/warpstreamlabs/bento/internal/httptransport"
 	"github.com/warpstreamlabs/bento/public/service"
 )
 
@@ -27,6 +28,7 @@ const (
 	hcFieldDumpRequestLogLevel = "dump_request_log_level"
 	hcFieldTLS                 = "tls"
 	hcFieldProxyURL            = "proxy_url"
+	hcFieldTransport           = "transport"
 )
 
 // ConfigField returns a public API config field spec for an HTTP component,
@@ -101,6 +103,7 @@ func ConfigField(defaultVerb string, forOutput bool, extraChildren ...*service.C
 			Description("An optional HTTP proxy URL.").
 			Advanced().
 			Optional(),
+		httptransport.CustomTransportConfigSpec(),
 	)
 
 	innerFields = append(innerFields, extraChildren...)
@@ -160,6 +163,10 @@ func ConfigFromParsed(pConf *service.ParsedConfig) (conf OldConfig, err error) {
 	if conf.clientCtor, err = oauth2ClientCtorFromParsed(pConf); err != nil {
 		return
 	}
+	if conf.transport, conf.customTransport, err = pConf.FieldHTTPTransport(hcFieldTransport); err != nil {
+		return
+	}
+
 	return
 }
 
@@ -184,4 +191,7 @@ type OldConfig struct {
 	ProxyURL            string
 	authSigner          func(f fs.FS, req *http.Request) error
 	clientCtor          func(context.Context, *http.Client) *http.Client
+
+	transport       *http.Transport
+	customTransport bool
 }
diff --git a/internal/httptransport/config.go b/internal/httptransport/config.go
@@ -0,0 +1,84 @@
+package httptransport
+
+import (
+	"time"
+
+	"github.com/warpstreamlabs/bento/public/service"
+)
+
+const (
+	hcFieldCustomTransportEnabled = "enabled"
+	hcFieldCustomTransport        = "custom_transport"
+	hcFieldDialContext            = "dial_context"
+	hcFieldDialContextTimeout     = "timeout"
+	hcFieldDialContextKeepAlive   = "keep_alive"
+	hcFieldForceAttemptHTTP2      = "force_http2"
+	hcFieldMaxIdleConns           = "max_idle_connections"
+	hcFieldIdleConnTimeout        = "idle_connection_timeout"
+	hcFieldTLSHandshakeTimeout    = "tls_handshake_timeout"
+	hcFieldExpectContinueTimeout  = "expect_continue_timeout"
+)
+
+func CustomTransportConfigSpec() *service.ConfigField {
+	return service.NewObjectField(hcFieldCustomTransport,
+		service.NewBoolField(hcFieldCustomTransportEnabled).
+			Description("Enables a custom HTTP transport. When set to `false` (default), Bento will use Go's [DefaultTransport](https://pkg.go.dev/net/http#DefaultTransport) for the underlying net/http transport. When enabled, settings from the `custom_transport` fields will be applied to the underlying transport. Note that other fields that modify the transport, such as TLS & ProxyURL, will always apply to the transport. The Env Var 'BENTO_OVERRIDE_DEFAULT_HTTP_TRANSPORT' can also be used to ensure the DefaultTransport isn't used.").
+			Advanced().
+			Version("1.13.0").
+			Default(false),
+		service.NewObjectField(hcFieldDialContext,
+			service.NewDurationField(hcFieldDialContextTimeout).
+				Description("Timeout for establishing new network connections.").
+				Advanced().
+				Version("1.13.0").
+				Default("30s"),
+			service.NewDurationField(hcFieldDialContextKeepAlive).
+				Description("Keep-alive period for an active network connections used by the dialer.").
+				Advanced().
+				Version("1.13.0").
+				Default("30s"),
+		).
+			Description("Settings for the dialer used to create new connections.").
+			Advanced().
+			Version("1.13.0").
+			Optional(),
+		service.NewBoolField(hcFieldForceAttemptHTTP2).
+			Description("If true, the transport will attempt to use HTTP/2.").
+			Advanced().
+			Version("1.13.0").
+			Default(true),
+		service.NewIntField(hcFieldMaxIdleConns).
+			Description("Controls the maximum number of idle (keep-alive) connections across all hosts. Zero means no limit.").
+			Advanced().
+			Version("1.13.0").
+			Default(100),
+		service.NewDurationField(hcFieldIdleConnTimeout).
+			Description("Maximum amount of time an idle (keep-alive) connection will remain idle before closing itself.").
+			Advanced().
+			Version("1.13.0").
+			Default("90s"),
+		service.NewDurationField(hcFieldTLSHandshakeTimeout).
+			Description("Maximum time allowed for the TLS handshake to complete when establishing connections.").
+			Advanced().
+			Version("1.13.0").
+			Default("10s"),
+		service.NewDurationField(hcFieldExpectContinueTimeout).
+			Description("Time to wait for a server's first response headers after fully writing the request headers if the request has an 'Expect: 100-continue' header. Zero means no timeout and causes the body to be sent immediately, without waiting for the server to approve.").
+			Advanced().
+			Version("1.13.0").
+			Default("1s"),
+	).
+		Description("Custom transport options.").
+		Advanced()
+}
+
+type CustomTransport struct {
+	CustomTransportEnabled bool
+	DialContextTimeout     time.Duration
+	DialContextKeepAlive   time.Duration
+	ForceAttemptHTTP2      bool
+	MaxIdleConns           int
+	IdleConnTimeout        time.Duration
+	TlsHandshakeTimeout    time.Duration
+	ExpectContinueTimeout  time.Duration
+}
diff --git a/public/service/config_http_transport.go b/public/service/config_http_transport.go
@@ -0,0 +1,86 @@
+package service
+
+import (
+	"net"
+	"net/http"
+	"os"
+)
+
+const (
+	fieldCustomTransportEnabled = "enabled"
+	fieldDialContext            = "dial_context"
+	fieldDialContextTimeout     = "timeout"
+	fieldDialContextKeepAlive   = "keep_alive"
+	fieldForceAttemptHTTP2      = "force_http2"
+	fieldMaxIdleConns           = "max_idle_connections"
+	fieldIdleConnTimeout        = "idle_connection_timeout"
+	fieldTLSHandshakeTimeout    = "tls_handshake_timeout"
+	fieldExpectContinueTimeout  = "expect_continue_timeout"
+)
+
+var overrideDefaultHTTPTransport bool = os.Getenv("BENTO_OVERRIDE_DEFAULT_HTTP_TRANSPORT") == "true"
+
+// FieldHTTPTransport constructs an *http.Transport based on configuration fields found at the given path.
+// It returns the transport, a boolean indicating if a custom transport is enabled, and an error if any configuration is invalid.
+// The default transport can be overridden by setting the BENTO_OVERRIDE_DEFAULT_HTTP_TRANSPORT environment variable to "true".
+func (pConf *ParsedConfig) FieldHTTPTransport(path ...string) (transport *http.Transport, customTransportEnabled bool, err error) {
+	tranPConf := pConf.Namespace(path...)
+
+	customTransportEnabled, err = tranPConf.FieldBool(fieldCustomTransportEnabled)
+	if err != nil {
+		return
+	}
+	if !customTransportEnabled && !overrideDefaultHTTPTransport {
+		return nil, false, nil
+	}
+
+	dialContextTimeout, err := tranPConf.FieldDuration([]string{fieldDialContext, fieldDialContextTimeout}...)
+	if err != nil {
+		return
+	}
+
+	dialContextKeepAlive, err := tranPConf.FieldDuration([]string{fieldDialContext, fieldDialContextKeepAlive}...)
+	if err != nil {
+		return
+	}
+
+	forceAttemptHTTP2, err := tranPConf.FieldBool(fieldForceAttemptHTTP2)
+	if err != nil {
+		return
+	}
+
+	maxIdleConns, err := tranPConf.FieldInt(fieldMaxIdleConns)
+	if err != nil {
+		return
+	}
+
+	idleConnTimeout, err := tranPConf.FieldDuration(fieldIdleConnTimeout)
+	if err != nil {
+		return
+	}
+
+	tlsHandshakeTimeout, err := tranPConf.FieldDuration(fieldTLSHandshakeTimeout)
+	if err != nil {
+		return
+	}
+
+	expectContinueTimeout, err := tranPConf.FieldDuration(fieldExpectContinueTimeout)
+	if err != nil {
+		return
+	}
+
+	transport = &http.Transport{
+		Proxy: http.ProxyFromEnvironment,
+		DialContext: (&net.Dialer{
+			Timeout:   dialContextTimeout,
+			KeepAlive: dialContextKeepAlive,
+		}).DialContext,
+		ForceAttemptHTTP2:     forceAttemptHTTP2,
+		MaxIdleConns:          maxIdleConns,
+		IdleConnTimeout:       idleConnTimeout,
+		TLSHandshakeTimeout:   tlsHandshakeTimeout,
+		ExpectContinueTimeout: expectContinueTimeout,
+	}
+
+	return transport, true, nil
+}
diff --git a/website/docs/components/inputs/http_client.md b/website/docs/components/inputs/http_client.md
@@ -102,6 +102,16 @@ input:
     drop_on: []
     successful_on: []
     proxy_url: "" # No default (optional)
+    custom_transport:
+      enabled: false
+      dial_context:
+        timeout: 30s
+        keep_alive: 30s
+      force_http2: true
+      max_idle_connections: 100
+      idle_connection_timeout: 90s
+      tls_handshake_timeout: 10s
+      expect_continue_timeout: 1s
     payload: "" # No default (optional)
     drop_empty_bodies: true
     stream:
@@ -724,6 +734,93 @@ An optional HTTP proxy URL.
 
 Type: `string`  
 
+### `custom_transport`
+
+Custom transport options.
+
+
+Type: `object`  
+
+### `custom_transport.enabled`
+
+Enables a custom HTTP transport. When set to `false` (default), Bento will use Go's [DefaultTransport](https://pkg.go.dev/net/http#DefaultTransport) for the underlying net/http transport. When enabled, settings from the `custom_transport` fields will be applied to the underlying transport. Note that other fields that modify the transport, such as TLS & ProxyURL, will always apply to the transport. The Env Var 'BENTO_OVERRIDE_DEFAULT_HTTP_TRANSPORT' can also be used to ensure the DefaultTransport isn't used.
+
+
+Type: `bool`  
+Default: `false`  
+Requires version 1.13.0 or newer  
+
+### `custom_transport.dial_context`
+
+Settings for the dialer used to create new connections.
+
+
+Type: `object`  
+Requires version 1.13.0 or newer  
+
+### `custom_transport.dial_context.timeout`
+
+Timeout for establishing new network connections.
+
+
+Type: `string`  
+Default: `"30s"`  
+Requires version 1.13.0 or newer  
+
+### `custom_transport.dial_context.keep_alive`
+
+Keep-alive period for an active network connections used by the dialer.
+
+
+Type: `string`  
+Default: `"30s"`  
+Requires version 1.13.0 or newer  
+
+### `custom_transport.force_http2`
+
+If true, the transport will attempt to use HTTP/2.
+
+
+Type: `bool`  
+Default: `true`  
+Requires version 1.13.0 or newer  
+
+### `custom_transport.max_idle_connections`
+
+Controls the maximum number of idle (keep-alive) connections across all hosts. Zero means no limit.
+
+
+Type: `int`  
+Default: `100`  
+Requires version 1.13.0 or newer  
+
+### `custom_transport.idle_connection_timeout`
+
+Maximum amount of time an idle (keep-alive) connection will remain idle before closing itself.
+
+
+Type: `string`  
+Default: `"90s"`  
+Requires version 1.13.0 or newer  
+
+### `custom_transport.tls_handshake_timeout`
+
+Maximum time allowed for the TLS handshake to complete when establishing connections.
+
+
+Type: `string`  
+Default: `"10s"`  
+Requires version 1.13.0 or newer  
+
+### `custom_transport.expect_continue_timeout`
+
+Time to wait for a server's first response headers after fully writing the request headers if the request has an 'Expect: 100-continue' header. Zero means no timeout and causes the body to be sent immediately, without waiting for the server to approve.
+
+
+Type: `string`  
+Default: `"1s"`  
+Requires version 1.13.0 or newer  
+
 ### `payload`
 
 An optional payload to deliver for each request.
diff --git a/website/docs/components/outputs/http_client.md b/website/docs/components/outputs/http_client.md
diff --git a/website/docs/components/processors/http.md b/website/docs/components/processors/http.md

Original file line number	Diff line number	Diff line change
`@@ -7,6 +7,7 @@ import (`
`7`	`7`	`"net/http"`
`8`	`8`	`"time"`
`9`	`9`
	`10`	`+ "github.com/warpstreamlabs/bento/internal/httptransport"`
`10`	`11`	`"github.com/warpstreamlabs/bento/public/service"`
`11`	`12`	`)`
`12`	`13`
`@@ -27,6 +28,7 @@ const (`
`27`	`28`	`hcFieldDumpRequestLogLevel = "dump_request_log_level"`
`28`	`29`	`hcFieldTLS = "tls"`
`29`	`30`	`hcFieldProxyURL = "proxy_url"`
	`31`	`+ hcFieldTransport = "transport"`
`30`	`32`	`)`
`31`	`33`
`32`	`34`	`// ConfigField returns a public API config field spec for an HTTP component,`
`@@ -101,6 +103,7 @@ func ConfigField(defaultVerb string, forOutput bool, extraChildren ...*service.C`
`101`	`103`	`Description("An optional HTTP proxy URL.").`
`102`	`104`	`Advanced().`
`103`	`105`	`Optional(),`
	`106`	`+ httptransport.CustomTransportConfigSpec(),`
`104`	`107`	`)`
`105`	`108`
`106`	`109`	`innerFields = append(innerFields, extraChildren...)`
`@@ -160,6 +163,10 @@ func ConfigFromParsed(pConf *service.ParsedConfig) (conf OldConfig, err error) {`
`160`	`163`	`if conf.clientCtor, err = oauth2ClientCtorFromParsed(pConf); err != nil {`
`161`	`164`	`return`
`162`	`165`	`}`
	`166`	`+ if conf.transport, conf.customTransport, err = pConf.FieldHTTPTransport(hcFieldTransport); err != nil {`
	`167`	`+ return`
	`168`	`+ }`
	`169`	`+`
`163`	`170`	`return`
`164`	`171`	`}`
`165`	`172`
`@@ -184,4 +191,7 @@ type OldConfig struct {`
`184`	`191`	`ProxyURL string`
`185`	`192`	`authSigner func(f fs.FS, req *http.Request) error`
`186`	`193`	`clientCtor func(context.Context, http.Client) http.Client`
	`194`	`+`
	`195`	`+ transport *http.Transport`
	`196`	`+ customTransport bool`
`187`	`197`	`}`