kafka: add socks5 proxy support for kafka output plugin

plugins/outputs/kafka/kafka.go

@@ -14,6 +14,7 @@ import (
 	tlsint "github.com/influxdata/telegraf/plugins/common/tls"
 	"github.com/influxdata/telegraf/plugins/outputs"
 	"github.com/influxdata/telegraf/plugins/serializers"
+	"golang.org/x/net/proxy"
 )
 
 var ValidTopicSuffixMethods = []string{
@@ -39,6 +40,11 @@ type (
 		MaxRetry         int         `toml:"max_retry"`
 		MaxMessageBytes  int         `toml:"max_message_bytes"`
 
+		Socks5ProxyEnabled bool      `toml:"socks5_enabled"`
+		Socks5ProxyAddress string    `toml:"socks5_address"`
+		Socks5ProxyUsername string   `toml:"socks5_username"`
+		Socks5ProxyPassword string   `toml:"socks5_password"`
+
 		Version string `toml:"version"`
 
 		// Legacy TLS config options
@@ -200,6 +206,12 @@ var sampleConfig = `
   ## Use TLS but skip chain & host verification
   # insecure_skip_verify = false
 
+  ## SOCKS5 proxy to use when connecting to brokers
+  socks5_enabled = false
+  socks5_address = "127.0.0.1:1080"
+  socks5_username = "alice"
+  socks5_password = "pass123"
+
   ## Optional SASL Config
   # sasl_username = "kafka"
   # sasl_password = "secret"
@@ -320,6 +332,23 @@ func (k *Kafka) Connect() error {
 		}
 	}
 
+	if k.Socks5ProxyEnabled {
+		config.Net.Proxy.Enable = true
+
+		var auth *proxy.Auth
+		if k.Socks5ProxyUsername != "" {
+			auth = new(proxy.Auth)
+			auth.User = k.Socks5ProxyUsername
+			auth.Password = k.Socks5ProxyPassword
+		}
+		dialer, err := proxy.SOCKS5("tcp", k.Socks5ProxyAddress, auth, proxy.Direct)
+		if err != nil {
+			log.Fatalf("Error while connecting to proxy server: %s", err)
+			return err
+		}
+		config.Net.Proxy.Dialer = dialer
+	}
+
 	if k.SASLUsername != "" && k.SASLPassword != "" {
 		config.Net.SASL.User = k.SASLUsername
 		config.Net.SASL.Password = k.SASLPassword