Add a header round tripper option to httpcommon

CHANGELOG.next.asciidoc

@@ -103,6 +103,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add state_job metricset to Kubernetes module{pull}26479[26479]
 - 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 User-Agent header to HTTP requests. {issue}18160[18160]
 
 *Packetbeat*
 

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/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/esleg/eslegclient/connection.go

@@ -25,6 +25,8 @@ import (
 	"io/ioutil"
 	"net/http"
 	"net/url"
+	"os"
+	"strings"
 	"time"
 
 	"go.elastic.co/apm/module/apmelasticsearch"
@@ -34,6 +36,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"
 )
@@ -110,6 +113,14 @@ func NewConnection(s ConnectionSettings) (*Connection, error) {
 		}
 	}
 
+	name, err := os.Executable()
+	if err != nil {
+		name = "ESLegClient"
+	} else {
+		name = strings.Title(name)
+	}
+	userAgent := useragent.UserAgent(name)
+
 	httpClient, err := s.Transport.Client(
 		httpcommon.WithLogger(logger),
 		httpcommon.WithIOStats(s.Observer),
@@ -119,6 +130,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

libbeat/kibana/client.go

@@ -29,12 +29,15 @@ import (
 	"net/http"
 	"net/textproto"
 	"net/url"
+	"os"
 	"path"
 	"strings"
 
 	"github.com/joeshaw/multierror"
 
 	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/common/transport/httpcommon"
+	"github.com/elastic/beats/v7/libbeat/common/useragent"
 	"github.com/elastic/beats/v7/libbeat/logp"
 )
 
@@ -143,7 +146,16 @@ func NewClientWithConfigDefault(config *ClientConfig, defaultPort int) (*Client,
 		headers.Set(k, v)
 	}
 
-	rt, err := config.Transport.Client()
+	name, err := os.Executable()
+	if err != nil {
+		log.Errorf("Unable to get running executable name: %v", err)
+		name = "KibanaClient"
+	} else {
+		name = strings.Title(name)
+	}
+	userAgent := useragent.UserAgent(name)
+
+	rt, err := config.Transport.Client(httpcommon.WithHeaderRoundTripper(map[string]string{"User-Agent": userAgent}))
 	if err != nil {
 		return nil, err
 	}

metricbeat/helper/http.go

@@ -29,10 +29,13 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/v7/libbeat/common/transport/httpcommon"
+	"github.com/elastic/beats/v7/libbeat/common/useragent"
 	"github.com/elastic/beats/v7/metricbeat/helper/dialer"
 	"github.com/elastic/beats/v7/metricbeat/mb"
 )
 
+var userAgent = useragent.UserAgent("Metricbeat")
+
 // HTTP is a custom HTTP Client that handle the complexity of connection and retrieving information
 // from HTTP endpoint.
 type HTTP struct {
@@ -87,6 +90,7 @@ func NewHTTPFromConfig(config Config, hostData mb.HostData) (*HTTP, error) {
 	client, err := config.Transport.Client(
 		httpcommon.WithBaseDialer(dialer),
 		httpcommon.WithAPMHTTPInstrumentation(),
+		httpcommon.WithHeaderRoundTripper(map[string]string{"User-Agent": userAgent}),
 	)
 	if err != nil {
 		return nil, err

metricbeat/helper/http_test.go

@@ -269,6 +269,31 @@ func TestOverUnixSocket(t *testing.T) {
 	}
 }
 
+func TestUserAgentCheck(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 := defaultConfig()
+	hostData := mb.HostData{
+		URI:          ts.URL,
+		SanitizedURI: ts.URL,
+	}
+
+	h, err := NewHTTPFromConfig(cfg, hostData)
+	require.NoError(t, err)
+
+	res, err := h.FetchResponse()
+	require.NoError(t, err)
+	res.Body.Close()
+
+	assert.Equal(t, http.StatusOK, res.StatusCode)
+	assert.Contains(t, ua, "Metricbeat")
+}
+
 func checkTimeout(t *testing.T, h *HTTP) {
 	t.Helper()
 

x-pack/filebeat/input/httpjson/input.go

@@ -162,6 +162,7 @@ func newHTTPClient(ctx context.Context, config config) (*http.Client, error) {
 		config.Transport.Client(
 			httpcommon.WithAPMHTTPInstrumentation(),
 			httpcommon.WithKeepaliveSettings{Disable: true},
+			httpcommon.WithHeaderRoundTripper(map[string]string{"User-Agent": userAgent}),
 		)
 	if err != nil {
 		return nil, err

x-pack/filebeat/input/httpjson/requester.go

@@ -194,7 +194,6 @@ func (r *requester) createHTTPRequest(ctx context.Context, ri *requestInfo) (*ht
 	req = req.WithContext(ctx)
 	req.Header.Set("Accept", "application/json")
 	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("User-Agent", userAgent)
 	if r.apiKey != "" {
 		if r.authScheme != "" {
 			req.Header.Set("Authorization", r.authScheme+" "+r.apiKey)