Add a header round tripper option to httpcommon (elastic#27509)

* Add a header round tripper option to httpcommon

Add a new TransportOption to add a set of headers to each HTTP request
through a custom http.RoundTripper. It will set the passed headers to
each request if the header key is not present. Use the new transport
option to add User-Agent headers to heartbeat, metricbeat, and filebeat.

* Fix useragent injection in heartbeat IPMonitor jobs

* Add User-Agent header to kibana and eslegclient

Add the User-Agent header with the RoundTripper to requests made by the
Kibana and eslegclient. The value for the User-Agent will be constructed
from what is returned by the os executable name.

* Fix syntax errors

* Get user agent from existing ingo

Get user-agent values for the Kibana and eslegclient from existing
config settings that get passed into the clients.

* change from settings.Name to b.Info.Name

* Fix missing param

* Rename export dashboard useragent value

* Review feedback

* Change beat.Info.Name to beat.Info.Beat

* Fix typo

CHANGELOG.next.asciidoc

@@ -107,6 +107,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Bump AWS SDK version to v0.24.0 for WebIdentity authentication flow {issue}19393[19393] {pull}27126[27126]
 - Add Linux pressure metricset {pull}27355[27355]
 - Add support for kube-state-metrics v2.0.0 {pull}27552[27552]
+- Add User-Agent header to HTTP requests. {issue}18160[18160] {pull}27509[27509]
 
 *Packetbeat*
 

dev-tools/cmd/dashboards/export_dashboards.go

@@ -78,7 +78,7 @@ func main() {
 		Path:      u.Path,
 		SpaceID:   *spaceID,
 		Transport: transport,
-	})
+	}, "Beat Development Tools")
 	if err != nil {
 		log.Fatalf("Error while connecting to Kibana: %v", err)
 	}

filebeat/beater/filebeat.go

@@ -174,7 +174,7 @@ func (fb *Filebeat) setupPipelineLoaderCallback(b *beat.Beat) error {
 
 	overwritePipelines := true
 	b.OverwritePipelinesCallback = func(esConfig *common.Config) error {
-		esClient, err := eslegclient.NewConnectedClient(esConfig)
+		esClient, err := eslegclient.NewConnectedClient(esConfig, "Filebeat")
 		if err != nil {
 			return err
 		}
@@ -428,7 +428,7 @@ func (fb *Filebeat) Stop() {
 // Create a new pipeline loader (es client) factory
 func newPipelineLoaderFactory(esConfig *common.Config) fileset.PipelineLoaderFactory {
 	pipelineLoaderFactory := func() (fileset.PipelineLoader, error) {
-		esClient, err := eslegclient.NewConnectedClient(esConfig)
+		esClient, err := eslegclient.NewConnectedClient(esConfig, "Filebeat")
 		if err != nil {
 			return nil, errors.Wrap(err, "Error creating Elasticsearch client")
 		}

heartbeat/monitors/active/http/http.go

@@ -29,6 +29,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/common/transport/httpcommon"
 	"github.com/elastic/beats/v7/libbeat/common/transport/tlscommon"
+	"github.com/elastic/beats/v7/libbeat/common/useragent"
 	"github.com/elastic/beats/v7/libbeat/logp"
 )
 
@@ -38,6 +39,8 @@ func init() {
 
 var debugf = logp.MakeDebug("http")
 
+var userAgent = useragent.UserAgent("Heartbeat")
+
 // Create makes a new HTTP monitor
 func create(
 	name string,
@@ -128,5 +131,6 @@ func newRoundTripper(config *Config) (http.RoundTripper, error) {
 		httpcommon.WithKeepaliveSettings{
 			Disable: true,
 		},
+		httpcommon.WithHeaderRoundTripper(map[string]string{"User-Agent": userAgent}),
 	)
 }

heartbeat/monitors/active/http/http_test.go

@@ -36,6 +36,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/elastic/beats/v7/heartbeat/hbtest"
@@ -674,3 +675,28 @@ func mustParseURL(t *testing.T, url string) *url.URL {
 	}
 	return parsed
 }
+
+func TestUserAgentInject(t *testing.T) {
+	ua := ""
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ua = r.Header.Get("User-Agent")
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer ts.Close()
+
+	cfg, err := common.NewConfigFrom(map[string]interface{}{
+		"urls": ts.URL,
+	})
+	require.NoError(t, err)
+
+	p, err := create("ua", cfg)
+	require.NoError(t, err)
+
+	sched, _ := schedule.Parse("@every 1s")
+	job := wrappers.WrapCommon(p.Jobs, stdfields.StdMonitorFields{ID: "test", Type: "http", Schedule: sched, Timeout: 1})[0]
+
+	event := &beat.Event{}
+	_, err = job(event)
+	require.NoError(t, err)
+	assert.Contains(t, ua, "Heartbeat")
+}

heartbeat/monitors/active/http/task.go

@@ -41,12 +41,10 @@ import (
 	"github.com/elastic/beats/v7/heartbeat/reason"
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/common/transport/httpcommon"
 	"github.com/elastic/beats/v7/libbeat/common/transport/tlscommon"
-	"github.com/elastic/beats/v7/libbeat/common/useragent"
 )
 
-var userAgent = useragent.UserAgent("Heartbeat")
-
 func newHTTPMonitorHostJob(
 	addr string,
 	config *Config,
@@ -141,27 +139,28 @@ func createPingFactory(
 		// prevents following redirects in this case, we know that
 		// config.MaxRedirects must be zero to even be here
 		checkRedirect := makeCheckRedirect(0, nil)
+		transport := &SimpleTransport{
+			Dialer: dialer,
+			OnStartWrite: func() {
+				cbMutex.Lock()
+				writeStart = time.Now()
+				cbMutex.Unlock()
+			},
+			OnEndWrite: func() {
+				cbMutex.Lock()
+				writeEnd = time.Now()
+				cbMutex.Unlock()
+			},
+			OnStartRead: func() {
+				cbMutex.Lock()
+				readStart = time.Now()
+				cbMutex.Unlock()
+			},
+		}
 		client := &http.Client{
 			CheckRedirect: checkRedirect,
 			Timeout:       timeout,
-			Transport: &SimpleTransport{
-				Dialer: dialer,
-				OnStartWrite: func() {
-					cbMutex.Lock()
-					writeStart = time.Now()
-					cbMutex.Unlock()
-				},
-				OnEndWrite: func() {
-					cbMutex.Lock()
-					writeEnd = time.Now()
-					cbMutex.Unlock()
-				},
-				OnStartRead: func() {
-					cbMutex.Lock()
-					readStart = time.Now()
-					cbMutex.Unlock()
-				},
-			},
+			Transport:     httpcommon.HeaderRoundTripper(transport, map[string]string{"User-Agent": userAgent}),
 		}
 
 		_, end, err := execPing(event, client, request, body, timeout, validator, config.Response)
@@ -206,9 +205,6 @@ func buildRequest(addr string, config *Config, enc contentEncoder) (*http.Reques
 
 		request.Header.Add(k, v)
 	}
-	if ua := request.Header.Get("User-Agent"); ua == "" {
-		request.Header.Set("User-Agent", userAgent)
-	}
 
 	if enc != nil {
 		enc.AddHeaders(&request.Header)

heartbeat/monitors/active/http/task_test.go

@@ -24,8 +24,6 @@ import (
 	"reflect"
 	"testing"
 
-	"github.com/elastic/beats/v7/libbeat/common/useragent"
-
 	"github.com/stretchr/testify/require"
 
 	"github.com/stretchr/testify/assert"
@@ -174,13 +172,6 @@ func TestRequestBuildingWithCustomHost(t *testing.T) {
 	}
 }
 
-func TestRequestBuildingWithNoUserAgent(t *testing.T) {
-	request, err := buildRequest("localhost", &Config{}, nilEncoder{})
-
-	require.NoError(t, err)
-	assert.Equal(t, useragent.UserAgent("Heartbeat"), request.Header.Get("User-Agent"))
-}
-
 func TestRequestBuildingWithExplicitUserAgent(t *testing.T) {
 	expectedUserAgent := "some-user-agent"
 

libbeat/cmd/export/dashboard.go

@@ -58,7 +58,7 @@ func GenDashboardCmd(settings instance.Settings) *cobra.Command {
 			// part of the initialization.
 			initConfig := instance.InitKibanaConfig(b.Config)
 
-			client, err := kibana.NewKibanaClient(initConfig)
+			client, err := kibana.NewKibanaClient(initConfig, b.Info.Beat)
 			if err != nil {
 				fatalf("Error creating Kibana client: %+v.\n", err)
 			}

libbeat/cmd/instance/beat.go

@@ -531,7 +531,7 @@ func (b *Beat) Setup(settings Settings, bt beat.Creator, setup SetupSettings) er
 			if outCfg.Name() != "elasticsearch" {
 				return fmt.Errorf("Index management requested but the Elasticsearch output is not configured/enabled")
 			}
-			esClient, err := eslegclient.NewConnectedClient(outCfg.Config())
+			esClient, err := eslegclient.NewConnectedClient(outCfg.Config(), b.Info.Beat)
 			if err != nil {
 				return err
 			}
@@ -808,7 +808,7 @@ func (b *Beat) loadDashboards(ctx context.Context, force bool) error {
 		// initKibanaConfig will attach the username and password into kibana config as a part of the initialization.
 		kibanaConfig := InitKibanaConfig(b.Config)
 
-		client, err := kibana.NewKibanaClient(kibanaConfig)
+		client, err := kibana.NewKibanaClient(kibanaConfig, b.Info.Beat)
 		if err != nil {
 			return fmt.Errorf("error connecting to Kibana: %v", err)
 		}

libbeat/common/transport/httpcommon/httpcommon.go

@@ -134,6 +134,20 @@ type extraOptionFunc func(*extraSettings)
 func (extraOptionFunc) sealTransportOption()           {}
 func (fn extraOptionFunc) applyExtra(s *extraSettings) { fn(s) }
 
+type headerRoundTripper struct {
+	headers map[string]string
+	rt      http.RoundTripper
+}
+
+func (rt *headerRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	for k, v := range rt.headers {
+		if len(req.Header.Get(k)) == 0 {
+			req.Header.Set(k, v)
+		}
+	}
+	return rt.rt.RoundTrip(req)
+}
+
 // DefaultHTTPTransportSettings returns the default HTTP transport setting.
 func DefaultHTTPTransportSettings() HTTPTransportSettings {
 	return HTTPTransportSettings{
@@ -373,6 +387,19 @@ func WithAPMHTTPInstrumentation() TransportOption {
 	return withAPMHTTPRountTripper
 }
 
+// HeaderRoundTripper will return a RoundTripper that sets header KVs if the key is not present.
+func HeaderRoundTripper(rt http.RoundTripper, headers map[string]string) http.RoundTripper {
+	return &headerRoundTripper{headers, rt}
+}
+
+// WithHeaderRoundTripper instuments the HTTP client via a custom http.RoundTripper.
+// This RoundTripper will add headers to each request if the key is not present.
+func WithHeaderRoundTripper(headers map[string]string) TransportOption {
+	return WithModRoundtripper(func(rt http.RoundTripper) http.RoundTripper {
+		return HeaderRoundTripper(rt, headers)
+	})
+}
+
 // WithLogger sets the internal logger that will be used to log dial or TCP level errors.
 // Logging at the connection level will only happen if the logger has been set.
 func WithLogger(logger *logp.Logger) TransportOption {

libbeat/dashboards/dashboards.go

@@ -54,13 +54,13 @@ func ImportDashboards(
 		return errors.New("kibana configuration missing for loading dashboards")
 	}
 
-	return setupAndImportDashboardsViaKibana(ctx, beatInfo.Hostname, kibanaConfig, &dashConfig, msgOutputter, pattern)
+	return setupAndImportDashboardsViaKibana(ctx, beatInfo.Hostname, beatInfo.Beat, kibanaConfig, &dashConfig, msgOutputter, pattern)
 }
 
-func setupAndImportDashboardsViaKibana(ctx context.Context, hostname string, kibanaConfig *common.Config,
+func setupAndImportDashboardsViaKibana(ctx context.Context, hostname, beatname string, kibanaConfig *common.Config,
 	dashboardsConfig *Config, msgOutputter MessageOutputter, fields common.MapStr) error {
 
-	kibanaLoader, err := NewKibanaLoader(ctx, kibanaConfig, dashboardsConfig, hostname, msgOutputter)
+	kibanaLoader, err := NewKibanaLoader(ctx, kibanaConfig, dashboardsConfig, hostname, msgOutputter, beatname)
 	if err != nil {
 		return fmt.Errorf("fail to create the Kibana loader: %v", err)
 	}

libbeat/dashboards/kibana_loader.go

@@ -49,13 +49,13 @@ type KibanaLoader struct {
 }
 
 // NewKibanaLoader creates a new loader to load Kibana files
-func NewKibanaLoader(ctx context.Context, cfg *common.Config, dashboardsConfig *Config, hostname string, msgOutputter MessageOutputter) (*KibanaLoader, error) {
+func NewKibanaLoader(ctx context.Context, cfg *common.Config, dashboardsConfig *Config, hostname string, msgOutputter MessageOutputter, beatname string) (*KibanaLoader, error) {
 
 	if cfg == nil || !cfg.Enabled() {
 		return nil, fmt.Errorf("Kibana is not configured or enabled")
 	}
 
-	client, err := getKibanaClient(ctx, cfg, dashboardsConfig.Retry, 0)
+	client, err := getKibanaClient(ctx, cfg, dashboardsConfig.Retry, 0, beatname)
 	if err != nil {
 		return nil, fmt.Errorf("Error creating Kibana client: %v", err)
 	}
@@ -76,15 +76,15 @@ func NewKibanaLoader(ctx context.Context, cfg *common.Config, dashboardsConfig *
 	return &loader, nil
 }
 
-func getKibanaClient(ctx context.Context, cfg *common.Config, retryCfg *Retry, retryAttempt uint) (*kibana.Client, error) {
-	client, err := kibana.NewKibanaClient(cfg)
+func getKibanaClient(ctx context.Context, cfg *common.Config, retryCfg *Retry, retryAttempt uint, beatname string) (*kibana.Client, error) {
+	client, err := kibana.NewKibanaClient(cfg, beatname)
 	if err != nil {
 		if retryCfg.Enabled && (retryCfg.Maximum == 0 || retryCfg.Maximum > retryAttempt) {
 			select {
 			case <-ctx.Done():
 				return nil, err
 			case <-time.After(retryCfg.Interval):
-				return getKibanaClient(ctx, cfg, retryCfg, retryAttempt+1)
+				return getKibanaClient(ctx, cfg, retryCfg, retryAttempt+1, beatname)
 			}
 		}
 		return nil, fmt.Errorf("Error creating Kibana client: %v", err)

libbeat/esleg/eslegclient/connection.go

@@ -34,6 +34,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/common/transport/httpcommon"
 	"github.com/elastic/beats/v7/libbeat/common/transport/kerberos"
 	"github.com/elastic/beats/v7/libbeat/common/transport/tlscommon"
+	"github.com/elastic/beats/v7/libbeat/common/useragent"
 	"github.com/elastic/beats/v7/libbeat/logp"
 	"github.com/elastic/beats/v7/libbeat/testing"
 )
@@ -57,7 +58,8 @@ type Connection struct {
 
 // ConnectionSettings are the settings needed for a Connection
 type ConnectionSettings struct {
-	URL string
+	URL      string
+	Beatname string
 
 	Username string
 	Password string
@@ -110,6 +112,11 @@ func NewConnection(s ConnectionSettings) (*Connection, error) {
 		}
 	}
 
+	if s.Beatname == "" {
+		s.Beatname = "Libbeat"
+	}
+	userAgent := useragent.UserAgent(s.Beatname)
+
 	httpClient, err := s.Transport.Client(
 		httpcommon.WithLogger(logger),
 		httpcommon.WithIOStats(s.Observer),
@@ -119,6 +126,7 @@ func NewConnection(s ConnectionSettings) (*Connection, error) {
 			// eg, like in https://github.com/elastic/apm-server/blob/7.7/elasticsearch/client.go
 			return apmelasticsearch.WrapRoundTripper(rt)
 		}),
+		httpcommon.WithHeaderRoundTripper(map[string]string{"User-Agent": userAgent}),
 	)
 	if err != nil {
 		return nil, err
@@ -160,7 +168,7 @@ func settingsWithDefaults(s ConnectionSettings) ConnectionSettings {
 // configuration. It accepts the same configuration parameters as the Elasticsearch
 // output, except for the output specific configuration options.  If multiple hosts
 // are defined in the configuration, a client is returned for each of them.
-func NewClients(cfg *common.Config) ([]Connection, error) {
+func NewClients(cfg *common.Config, beatname string) ([]Connection, error) {
 	config := defaultConfig()
 	if err := cfg.Unpack(&config); err != nil {
 		return nil, err
@@ -185,6 +193,7 @@ func NewClients(cfg *common.Config) ([]Connection, error) {
 
 		client, err := NewConnection(ConnectionSettings{
 			URL:              esURL,
+			Beatname:         beatname,
 			Kerberos:         config.Kerberos,
 			Username:         config.Username,
 			Password:         config.Password,
@@ -205,8 +214,8 @@ func NewClients(cfg *common.Config) ([]Connection, error) {
 	return clients, nil
 }
 
-func NewConnectedClient(cfg *common.Config) (*Connection, error) {
-	clients, err := NewClients(cfg)
+func NewConnectedClient(cfg *common.Config, beatname string) (*Connection, error) {
+	clients, err := NewClients(cfg, beatname)
 	if err != nil {
 		return nil, err
 	}