Custom HTTP headers for the Elasticsearch output (#3400)

Configuration looks like this:

```
output.elasticsearch.headers:
  X-My-Header: Contents of the header
```

To use from the CLI:

```
metricbeat -E "output.elasticsearch.headers.X-test=Test value"
```

It's not possible to set the same header name more than once with different values,
but it is possible to separate header values with a comma, which has the same meaning
as per the RFC.

Closes #1768.

CHANGELOG.asciidoc

@@ -63,6 +63,7 @@ https://github.com/elastic/beats/compare/v5.1.1...master[Check the HEAD diff]
 - Added a NOTICE file containing the notices and licenses of the dependencies. {pull}3334[3334].
 - Files created by Beats (logs, registry, file output) will have 0600 permissions. {pull}3387[3387].
 - RPM/deb packages will now install the config file with 0600 permissions. {pull}3382[3382]
+- Add the option to pass custom HTTP headers to the Elasticsearch output. {pull}3400[3400]
 
 *Metricbeat*
 

filebeat/filebeat.full.yml

@@ -355,6 +355,10 @@ output.elasticsearch:
   # Optional HTTP Path
   #path: "/elasticsearch"
 
+  # Custom HTTP headers to add to each request
+  #headers:
+  #  X-My-Header: Contents of the header
+
   # Proxy server url
   #proxy_url: http://proxy:3128
 

heartbeat/heartbeat.full.yml

@@ -310,6 +310,10 @@ output.elasticsearch:
   # Optional HTTP Path
   #path: "/elasticsearch"
 
+  # Custom HTTP headers to add to each request
+  #headers:
+  #  X-My-Header: Contents of the header
+
   # Proxy server url
   #proxy_url: http://proxy:3128
 

libbeat/_meta/config.full.yml

@@ -112,6 +112,10 @@ output.elasticsearch:
   # Optional HTTP Path
   #path: "/elasticsearch"
 
+  # Custom HTTP headers to add to each request
+  #headers:
+  #  X-My-Header: Contents of the header
+
   # Proxy server url
   #proxy_url: http://proxy:3128
 

libbeat/docs/outputconfig.asciidoc

@@ -135,6 +135,20 @@ An HTTP path prefix that is prepended to the HTTP API calls. This is useful for
 the cases where Elasticsearch listens behind an HTTP reverse proxy that exports
 the API under a custom prefix.
 
+===== headers
+
+Custom HTTP headers to add to each request created by the Elasticsearch output.
+Example:
+
+[source,yaml]
+------------------------------------------------------------------------------
+output.elasticsearch.headers:
+  X-My-Header: Header contents
+------------------------------------------------------------------------------
+
+It is generally possible to specify multiple header values for the same header
+name by separating them with a comma.
+
 ===== proxy_url
 
 The URL of the proxy to use when connecting to the Elasticsearch servers. The

libbeat/outputs/elasticsearch/client.go

@@ -46,6 +46,7 @@ type ClientSettings struct {
 	TLS                *transport.TLSConfig
 	Username, Password string
 	Parameters         map[string]string
+	Headers            map[string]string
 	Index              outil.Selector
 	Pipeline           *outil.Selector
 	Timeout            time.Duration
@@ -58,6 +59,7 @@ type Connection struct {
 	URL      string
 	Username string
 	Password string
+	Headers  map[string]string
 
 	http              *http.Client
 	onConnectCallback func() error
@@ -160,6 +162,7 @@ func NewClient(
 			URL:      s.URL,
 			Username: s.Username,
 			Password: s.Password,
+			Headers:  s.Headers,
 			http: &http.Client{
 				Transport: &http.Transport{
 					Dial:    dialer.Dial,
@@ -208,6 +211,7 @@ func (client *Client) Clone() *Client {
 			Username:         client.Username,
 			Password:         client.Password,
 			Parameters:       nil, // XXX: do not pass params?
+			Headers:          client.Headers,
 			Timeout:          client.http.Timeout,
 			CompressionLevel: client.compressionLevel,
 		},
@@ -700,6 +704,10 @@ func (conn *Connection) execHTTPRequest(req *http.Request) (int, []byte, error)
 		req.SetBasicAuth(conn.Username, conn.Password)
 	}
 
+	for name, value := range conn.Headers {
+		req.Header.Add(name, value)
+	}
+
 	resp, err := conn.http.Do(req)
 	if err != nil {
 		return 0, nil, err

libbeat/outputs/elasticsearch/client_test.go

@@ -4,6 +4,8 @@ package elasticsearch
 
 import (
 	"fmt"
+	"net/http"
+	"net/http/httptest"
 	"strings"
 	"testing"
 	"time"
@@ -271,3 +273,38 @@ func BenchmarkCollectPublishFailAll(b *testing.B) {
 		}
 	}
 }
+
+func TestClientWithHeaders(t *testing.T) {
+	requestCount := 0
+	// start a mock HTTP server
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "testing value", r.Header.Get("X-Test"))
+		requestCount += 1
+		fmt.Fprintln(w, "Hello, client")
+	}))
+	defer ts.Close()
+
+	client, err := NewClient(ClientSettings{
+		URL:   ts.URL,
+		Index: outil.MakeSelector(outil.ConstSelectorExpr("test")),
+		Headers: map[string]string{
+			"X-Test": "testing value",
+		},
+	}, nil)
+	assert.NoError(t, err)
+
+	// simple ping
+	client.Ping(1 * time.Second)
+	assert.Equal(t, 1, requestCount)
+
+	// bulk request
+	event := outputs.Data{Event: common.MapStr{
+		"@timestamp": common.Time(time.Now()),
+		"type":       "libbeat",
+		"message":    "Test message from libbeat",
+	}}
+	events := []outputs.Data{event, event, event}
+	_, err = client.PublishEvents(events)
+	assert.NoError(t, err)
+	assert.Equal(t, 2, requestCount)
+}

libbeat/outputs/elasticsearch/config.go

@@ -10,6 +10,7 @@ type elasticsearchConfig struct {
 	Protocol         string             `config:"protocol"`
 	Path             string             `config:"path"`
 	Params           map[string]string  `config:"parameters"`
+	Headers          map[string]string  `config:"headers"`
 	Username         string             `config:"username"`
 	Password         string             `config:"password"`
 	ProxyURL         string             `config:"proxy_url"`

libbeat/outputs/elasticsearch/output.go

@@ -144,6 +144,7 @@ func NewElasticsearchClients(cfg *common.Config) ([]Client, error) {
 			Username:         config.Username,
 			Password:         config.Password,
 			Parameters:       params,
+			Headers:          config.Headers,
 			Timeout:          config.Timeout,
 			CompressionLevel: config.CompressionLevel,
 		}, nil)
@@ -368,6 +369,7 @@ func makeClientFactory(
 			Username:         config.Username,
 			Password:         config.Password,
 			Parameters:       params,
+			Headers:          config.Headers,
 			Timeout:          config.Timeout,
 			CompressionLevel: config.CompressionLevel,
 		}, onConnected)

metricbeat/metricbeat.full.yml

@@ -406,6 +406,10 @@ output.elasticsearch:
   # Optional HTTP Path
   #path: "/elasticsearch"
 
+  # Custom HTTP headers to add to each request
+  #headers:
+  #  X-My-Header: Contents of the header
+
   # Proxy server url
   #proxy_url: http://proxy:3128
 

packetbeat/packetbeat.full.yml

@@ -566,6 +566,10 @@ output.elasticsearch:
   # Optional HTTP Path
   #path: "/elasticsearch"
 
+  # Custom HTTP headers to add to each request
+  #headers:
+  #  X-My-Header: Contents of the header
+
   # Proxy server url
   #proxy_url: http://proxy:3128
 

winlogbeat/winlogbeat.full.yml

@@ -147,6 +147,10 @@ output.elasticsearch:
   # Optional HTTP Path
   #path: "/elasticsearch"
 
+  # Custom HTTP headers to add to each request
+  #headers:
+  #  X-My-Header: Contents of the header
+
   # Proxy server url
   #proxy_url: http://proxy:3128