diff --git a/plugins/outputs/influxdb/README.md b/plugins/outputs/influxdb/README.md index 10ebde2646e38..478af82f0608d 100644 --- a/plugins/outputs/influxdb/README.md +++ b/plugins/outputs/influxdb/README.md @@ -34,6 +34,10 @@ to use them. # urls = ["udp://127.0.0.1:8089"] # urls = ["http://127.0.0.1:8086"] + ## Local address to bind when connecting to the server + ## If empty or not set, the local address is automatically chosen. + # local_address = "" + ## The target database for metrics; will be created as needed. ## For UDP url endpoint database needs to be configured on server side. # database = "telegraf" diff --git a/plugins/outputs/influxdb/http.go b/plugins/outputs/influxdb/http.go index 22fd9a8e5256b..4120f861250f8 100644 --- a/plugins/outputs/influxdb/http.go +++ b/plugins/outputs/influxdb/http.go @@ -87,6 +87,7 @@ func (r WriteResponseError) Error() string { type HTTPConfig struct { URL *url.URL + LocalAddr *net.TCPAddr UserAgent string Timeout time.Duration Username config.Secret @@ -164,9 +165,15 @@ func NewHTTPClient(cfg HTTPConfig) (*httpClient, error) { var transport *http.Transport switch cfg.URL.Scheme { case "http", "https": + var dialerFunc func(ctx context.Context, network, addr string) (net.Conn, error) + if cfg.LocalAddr != nil { + dialer := &net.Dialer{LocalAddr: cfg.LocalAddr} + dialerFunc = dialer.DialContext + } transport = &http.Transport{ Proxy: proxy, TLSClientConfig: cfg.TLSConfig, + DialContext: dialerFunc, } case "unix": transport = &http.Transport{ diff --git a/plugins/outputs/influxdb/influxdb.go b/plugins/outputs/influxdb/influxdb.go index 8595a22b94260..fb909511294fc 100644 --- a/plugins/outputs/influxdb/influxdb.go +++ b/plugins/outputs/influxdb/influxdb.go @@ -7,7 +7,10 @@ import ( "errors" "fmt" "math/rand" + "net" "net/url" + "strconv" + "strings" "time" "github.com/influxdata/telegraf" @@ -37,6 +40,7 @@ type Client interface { // InfluxDB struct is the primary data structure for the plugin type InfluxDB struct { URLs []string `toml:"urls"` + LocalAddr string `toml:"local_address"` Username config.Secret `toml:"username"` Password config.Secret `toml:"password"` Database string `toml:"database"` @@ -89,16 +93,54 @@ func (i *InfluxDB) Connect() error { } } + var localIP *net.IPAddr + var localPort int + if i.LocalAddr != "" { + var err error + // Resolve the local address into IP address and the given port if any + addr, sPort, err := net.SplitHostPort(i.LocalAddr) + if err != nil { + if !strings.Contains(err.Error(), "missing port") { + return fmt.Errorf("invalid local address: %w", err) + } + addr = i.LocalAddr + } + localIP, err = net.ResolveIPAddr("ip", addr) + if err != nil { + return fmt.Errorf("cannot resolve local address: %w", err) + } + + if sPort != "" { + p, err := strconv.ParseUint(sPort, 10, 16) + if err != nil { + return fmt.Errorf("invalid port: %w", err) + } + localPort = int(p) + } + } + switch parts.Scheme { case "udp", "udp4", "udp6": - c, err := i.udpClient(parts) + var c Client + var err error + if i.LocalAddr == "" { + c, err = i.udpClient(parts, nil) + } else { + c, err = i.udpClient(parts, &net.UDPAddr{IP: localIP.IP, Port: localPort, Zone: localIP.Zone}) + } if err != nil { return err } i.clients = append(i.clients, c) case "http", "https", "unix": - c, err := i.httpClient(ctx, parts, proxy) + var c Client + var err error + if i.LocalAddr == "" { + c, err = i.httpClient(ctx, parts, nil, proxy) + } else { + c, err = i.httpClient(ctx, parts, &net.TCPAddr{IP: localIP.IP, Port: localPort, Zone: localIP.Zone}, proxy) + } if err != nil { return err } @@ -159,7 +201,7 @@ func (i *InfluxDB) Write(metrics []telegraf.Metric) error { return errors.New("could not write any address") } -func (i *InfluxDB) udpClient(address *url.URL) (Client, error) { +func (i *InfluxDB) udpClient(address *url.URL, localAddr *net.UDPAddr) (Client, error) { serializer := &influx.Serializer{UintSupport: i.InfluxUintSupport} if err := serializer.Init(); err != nil { return nil, err @@ -167,6 +209,7 @@ func (i *InfluxDB) udpClient(address *url.URL) (Client, error) { udpConfig := &UDPConfig{ URL: address, + LocalAddr: localAddr, MaxPayloadSize: int(i.UDPPayload), Serializer: serializer, Log: i.Log, @@ -180,7 +223,7 @@ func (i *InfluxDB) udpClient(address *url.URL) (Client, error) { return c, nil } -func (i *InfluxDB) httpClient(ctx context.Context, address *url.URL, proxy *url.URL) (Client, error) { +func (i *InfluxDB) httpClient(ctx context.Context, address *url.URL, localAddr *net.TCPAddr, proxy *url.URL) (Client, error) { tlsConfig, err := i.ClientConfig.TLSConfig() if err != nil { return nil, err @@ -193,6 +236,7 @@ func (i *InfluxDB) httpClient(ctx context.Context, address *url.URL, proxy *url. httpConfig := &HTTPConfig{ URL: address, + LocalAddr: localAddr, Timeout: time.Duration(i.Timeout), TLSConfig: tlsConfig, UserAgent: i.UserAgent, diff --git a/plugins/outputs/influxdb/influxdb_test.go b/plugins/outputs/influxdb/influxdb_test.go index 29c5c3c973f70..53e08c3b563f1 100644 --- a/plugins/outputs/influxdb/influxdb_test.go +++ b/plugins/outputs/influxdb/influxdb_test.go @@ -214,3 +214,23 @@ func TestWriteRecreateDatabaseIfDatabaseNotFound(t *testing.T) { // We only have one URL, so we expect an error require.Error(t, err) } + +func TestInfluxDBLocalAddress(t *testing.T) { + output := influxdb.InfluxDB{ + URLs: []string{"http://localhost:8086"}, + LocalAddr: "localhost", + + CreateHTTPClientF: func(_ *influxdb.HTTPConfig) (influxdb.Client, error) { + return &MockClient{ + DatabaseF: func() string { + return "telegraf" + }, + CreateDatabaseF: func() error { + return nil + }, + }, nil + }, + } + + require.NoError(t, output.Connect()) +} diff --git a/plugins/outputs/influxdb/sample.conf b/plugins/outputs/influxdb/sample.conf index fa23d68c5cc0b..1338a140586cc 100644 --- a/plugins/outputs/influxdb/sample.conf +++ b/plugins/outputs/influxdb/sample.conf @@ -8,6 +8,10 @@ # urls = ["udp://127.0.0.1:8089"] # urls = ["http://127.0.0.1:8086"] + ## Local address to bind when connecting to the server + ## If empty or not set, the local address is automatically chosen. + # local_address = "" + ## The target database for metrics; will be created as needed. ## For UDP url endpoint database needs to be configured on server side. # database = "telegraf" diff --git a/plugins/outputs/influxdb/udp.go b/plugins/outputs/influxdb/udp.go index 1d4fcf804db2e..0df2909ff3892 100644 --- a/plugins/outputs/influxdb/udp.go +++ b/plugins/outputs/influxdb/udp.go @@ -29,6 +29,7 @@ type Conn interface { type UDPConfig struct { MaxPayloadSize int URL *url.URL + LocalAddr *net.UDPAddr Serializer *influx.Serializer Dialer Dialer Log telegraf.Logger @@ -55,7 +56,7 @@ func NewUDPClient(config UDPConfig) (*udpClient, error) { dialer := config.Dialer if dialer == nil { - dialer = &netDialer{net.Dialer{}} + dialer = &netDialer{net.Dialer{LocalAddr: config.LocalAddr}} } client := &udpClient{