Amazon Elasticsearch support

libbeat/outputs/elasticsearch/aws.go

@@ -0,0 +1,82 @@
+package elasticsearch
+
+import (
+	"net/http"
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/defaults"
+	"github.com/aws/aws-sdk-go/aws/signer/v4"
+	"io/ioutil"
+	"bytes"
+	"github.com/elastic/beats/libbeat/logp"
+	"time"
+	"io"
+	"strings"
+)
+
+type AwsData struct {
+	Config aws.Config
+	Signer v4.Signer
+}
+
+// Calculate signature for request and append headers
+func (conn *Connection) performAwsSignature(req *http.Request) {
+
+	if conn.Aws == nil {
+		return
+	}
+
+	var reader io.ReadSeeker
+	if req.Body != nil {
+		body := req.Body
+		buf, err := ioutil.ReadAll(body)
+		if err != nil {
+			logp.Err("Error performing AWS signature: %s", err)
+			return
+		}
+		reader = bytes.NewReader(buf)
+		req.Body = ioutil.NopCloser(reader)
+	} else {
+		reader = nil
+	}
+
+	// AWS-SDK creates invalid signature when host contains a port
+	// See https://github.com/aws/aws-sdk-go/issues/1537
+	// Host manipulations can be removed when it's fixed
+	host := req.Host
+	colon := strings.IndexByte(req.Host, ':')
+	if colon >= 0 {
+		req.Host = req.Host[:colon]
+	}
+	_, err := conn.Aws.Signer.Sign(req, reader, "es", *conn.Aws.Config.Region, time.Now())
+	if err != nil {
+		logp.Err("Error performing AWS signature: %s", err)
+	}
+	req.Host = host
+}
+
+func (c *elasticsearchConfig) Aws() *AwsData {
+	if c.AwsConfig.Enabled {
+		credentials := defaults.CredChain(defaults.Config(), defaults.Handlers())
+		if _, err := credentials.Get(); err != nil {
+			logp.Err("Error loading AWS credentials: %s", err)
+			return nil
+		}
+
+		region := c.AwsConfig.Region
+		if region == "" {
+			region = "us-east-1"
+		}
+
+		config := aws.NewConfig().WithCredentials(credentials).WithRegion(region)
+		return &AwsData{
+			Config: *config,
+			Signer: v4.Signer{
+				Credentials: config.Credentials,
+				Logger:      aws.NewDefaultLogger(), // TODO remove debug logging
+				Debug:       aws.LogDebugWithSigning,
+			},
+		}
+	}
+
+	return nil
+}

libbeat/outputs/elasticsearch/client.go

@@ -56,6 +56,7 @@ type ClientSettings struct {
 	Timeout            time.Duration
 	CompressionLevel   int
 	Stats              *outputs.Stats
+	Aws                *AwsData
 }
 
 type connectCallback func(client *Client) error
@@ -72,6 +73,7 @@ type Connection struct {
 
 	encoder bodyEncoder
 	version string
+	Aws     *AwsData
 }
 
 var (
@@ -168,6 +170,7 @@ func NewClient(
 				Timeout: s.Timeout,
 			},
 			encoder: encoder,
+			Aws:     s.Aws,
 		},
 		tlsConfig: s.TLS,
 		index:     s.Index,
@@ -219,6 +222,7 @@ func (client *Client) Clone() *Client {
 			Headers:          client.Headers,
 			Timeout:          client.http.Timeout,
 			CompressionLevel: client.compressionLevel,
+			Aws:              client.Aws,
 		},
 		nil, // XXX: do not pass connection callback?
 	)
@@ -721,6 +725,10 @@ func (conn *Connection) execHTTPRequest(req *http.Request) (int, []byte, error)
 		req.Host = host
 	}
 
+	if conn.Aws != nil {
+		conn.performAwsSignature(req)
+	}
+
 	resp, err := conn.http.Do(req)
 	if err != nil {
 		return 0, nil, err

libbeat/outputs/elasticsearch/config.go

@@ -21,6 +21,12 @@ type elasticsearchConfig struct {
 	MaxRetries       int                `config:"max_retries"`
 	Timeout          time.Duration      `config:"timeout"`
 	Backoff          Backoff            `config:"backoff"`
+	AwsConfig        AwsConfig          `config:"aws"`
+}
+
+type AwsConfig struct {
+	Enabled bool   `config:"enabled"`
+	Region  string `config:"region"`
 }
 
 type Backoff struct {
@@ -49,6 +55,10 @@ var (
 			Init: 1 * time.Second,
 			Max:  60 * time.Second,
 		},
+		AwsConfig: AwsConfig{
+			Enabled: false,
+			Region:  "",
+		},
 	}
 )
 

libbeat/outputs/elasticsearch/elasticsearch.go

@@ -136,6 +136,7 @@ func makeES(
 			Timeout:          config.Timeout,
 			CompressionLevel: config.CompressionLevel,
 			Stats:            stats,
+			Aws:              config.Aws(),
 		}, &connectCallbackRegistry)
 		if err != nil {
 			return outputs.Fail(err)