feat(outputs.amqp): Add proxy support (#11649)

plugins/outputs/amqp/README.md

@@ -90,6 +90,10 @@ For an introduction to AMQP see:
   ## Use TLS but skip chain & host verification
   # insecure_skip_verify = false
 
+  ## Optional Proxy Configuration
+  # use_proxy = false
+  # proxy_url = "localhost:8888"
+
   ## If true use batch serialization format instead of line based delimiting.
   ## Only applies to data formats which are not line based such as JSON.
   ## Recommended to set to true.
@@ -120,3 +124,10 @@ Exchange types that do not use a routing key, `direct` and `header`, always use
 the empty string as the routing key.
 
 Metrics are published in batches based on the final routing key.
+
+### Proxy
+
+If you want to use a proxy, you need to set `use_proxy = true`. This will
+use the system's proxy settings to determine the proxy URL. If you need to
+specify a proxy URL manually, you can do so by using `proxy_url`, overriding
+the system settings.

plugins/outputs/amqp/amqp.go

@@ -12,12 +12,14 @@ import (
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/config"
 	"github.com/influxdata/telegraf/internal"
+	"github.com/influxdata/telegraf/plugins/common/proxy"
 	"github.com/influxdata/telegraf/plugins/common/tls"
 	"github.com/influxdata/telegraf/plugins/outputs"
 	"github.com/influxdata/telegraf/plugins/serializers"
 )
 
 // DO NOT REMOVE THE NEXT TWO LINES! This is required to embed the sampleConfig data.
+//
 //go:embed sample.conf
 var sampleConfig string
 
@@ -63,6 +65,7 @@ type AMQP struct {
 	ContentEncoding    string            `toml:"content_encoding"`
 	Log                telegraf.Logger   `toml:"-"`
 	tls.ClientConfig
+	proxy.TCPProxy
 
 	serializer   serializers.Serializer
 	connect      func(*ClientConfig) (Client, error)
@@ -283,6 +286,12 @@ func (q *AMQP) makeClientConfig() (*ClientConfig, error) {
 	}
 	clientConfig.tlsConfig = tlsConfig
 
+	dialer, err := q.TCPProxy.Proxy()
+	if err != nil {
+		return nil, err
+	}
+	clientConfig.dialer = dialer
+
 	var auth []amqp.Authentication
 	if strings.ToUpper(q.AuthMethod) == "EXTERNAL" {
 		auth = []amqp.Authentication{&externalAuth{}}

plugins/outputs/amqp/client.go

@@ -11,6 +11,7 @@ import (
 	amqp "github.com/rabbitmq/amqp091-go"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/common/proxy"
 )
 
 type ClientConfig struct {
@@ -26,6 +27,7 @@ type ClientConfig struct {
 	tlsConfig         *tls.Config
 	timeout           time.Duration
 	auth              []amqp.Authentication
+	dialer            *proxy.ProxiedDialer
 	log               telegraf.Logger
 }
 
@@ -50,7 +52,7 @@ func newClient(config *ClientConfig) (*client, error) {
 				TLSClientConfig: config.tlsConfig,
 				SASL:            config.auth, // if nil, it will be PLAIN taken from url
 				Dial: func(network, addr string) (net.Conn, error) {
-					return net.DialTimeout(network, addr, config.timeout)
+					return config.dialer.DialTimeout(network, addr, config.timeout)
 				},
 			})
 		if err == nil {

plugins/outputs/amqp/sample.conf

@@ -75,6 +75,10 @@
   ## Use TLS but skip chain & host verification
   # insecure_skip_verify = false
 
+  ## Optional Proxy Configuration
+  # use_proxy = false
+  # proxy_url = "localhost:8888"
+
   ## If true use batch serialization format instead of line based delimiting.
   ## Only applies to data formats which are not line based such as JSON.
   ## Recommended to set to true.