Update according to upstream api changes

raboof · raboof · commit 3e5da548bb52 · 2016-09-02T10:38:51.000+02:00
diff --git a/client.go b/client.go
@@ -1,7 +1,6 @@
 package http
 
 import (
-	"crypto/tls"
 	"expvar"
 	"fmt"
 	"io"
@@ -12,18 +11,32 @@ import (
 
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/libbeat/outputs"
+	"github.com/elastic/beats/libbeat/outputs/outil"
 	"github.com/elastic/beats/libbeat/outputs/transport"
 )
 
 type Client struct {
 	Connection
-	params map[string]string
+	tlsConfig *transport.TLSConfig
+	params    map[string]string
 
 	// additional configs
 	compressionLevel int
 	proxyURL         *url.URL
 }
 
+type ClientSettings struct {
+	URL                string
+	Proxy              *url.URL
+	TLS                *transport.TLSConfig
+	Username, Password string
+	Parameters         map[string]string
+	Index              outil.Selector
+	Pipeline           *outil.Selector
+	Timeout            time.Duration
+	CompressionLevel   int
+}
+
 type Connection struct {
 	URL      string
 	Username string
@@ -48,29 +61,38 @@ var (
 )
 
 func NewClient(
-	hostURL string, proxyURL *url.URL, tls *tls.Config,
-	username, password string,
-	params map[string]string,
-	timeout time.Duration,
-	compression int,
+	s ClientSettings,
 ) (*Client, error) {
 	proxy := http.ProxyFromEnvironment
-	if proxyURL != nil {
-		proxy = http.ProxyURL(proxyURL)
+	if s.Proxy != nil {
+		proxy = http.ProxyURL(s.Proxy)
 	}
 
-	logp.Info("Http url: %s", hostURL)
+	logp.Info("Http url: %s", s.URL)
 
-	dialer := transport.NetDialer(timeout)
-	dialer = transport.StatsDialer(dialer, &transport.IOStats{
+	// TODO: add socks5 proxy support
+	var dialer, tlsDialer transport.Dialer
+	var err error
+
+	dialer = transport.NetDialer(s.Timeout)
+	tlsDialer, err = transport.TLSDialer(dialer, s.TLS, s.Timeout)
+	if err != nil {
+		return nil, err
+	}
+
+	iostats := &transport.IOStats{
 		Read:        statReadBytes,
 		Write:       statWriteBytes,
 		ReadErrors:  statReadErrors,
 		WriteErrors: statWriteErrors,
-	})
+	}
+	dialer = transport.StatsDialer(dialer, iostats)
+	tlsDialer = transport.StatsDialer(tlsDialer, iostats)
+
+	params := s.Parameters
 
-	var err error
 	var encoder bodyEncoder
+	compression := s.CompressionLevel
 	if compression == 0 {
 		encoder = newJSONEncoder(nil)
 	} else {
@@ -82,22 +104,23 @@ func NewClient(
 
 	client := &Client{
 		Connection: Connection{
-			URL:      hostURL,
-			Username: username,
-			Password: password,
+			URL:      s.URL,
+			Username: s.Username,
+			Password: s.Password,
 			http: &http.Client{
 				Transport: &http.Transport{
-					Dial:            dialer.Dial,
-					TLSClientConfig: tls,
-					Proxy:           proxy,
+					Dial:    dialer.Dial,
+					DialTLS: tlsDialer.Dial,
+					Proxy:   proxy,
 				},
-				Timeout: timeout,
+				Timeout: s.Timeout,
 			},
 			encoder: encoder,
 		},
 		params: params,
 
-		proxyURL: proxyURL,
+		compressionLevel: compression,
+		proxyURL:         s.Proxy,
 	}
 
 	return client, nil
@@ -108,17 +131,17 @@ func (client *Client) Clone() *Client {
 	// client's close is for example generated for topology-map support. With params
 	// most likely containing the ingest node pipeline and default callback trying to
 	// create install a template, we don't want these to be included in the clone.
-
-	transport := client.http.Transport.(*http.Transport)
 	c, _ := NewClient(
-		client.URL,
-		client.proxyURL,
-		transport.TLSClientConfig,
-		client.Username,
-		client.Password,
-		nil, // XXX: do not pass params?
-		client.http.Timeout,
-		client.compressionLevel,
+		ClientSettings{
+			URL:              client.URL,
+			Proxy:            client.proxyURL,
+			TLS:              client.tlsConfig,
+			Username:         client.Username,
+			Password:         client.Password,
+			Parameters:       nil, // XXX: do not pass params?
+			Timeout:          client.http.Timeout,
+			CompressionLevel: client.compressionLevel,
+		},
 	)
 	return c
 }
diff --git a/http.go b/http.go
@@ -1,7 +1,6 @@
 package http
 
 import (
-	"crypto/tls"
 	"errors"
 	"net/url"
 	"strings"
@@ -13,6 +12,7 @@ import (
 	"github.com/elastic/beats/libbeat/outputs"
 	"github.com/elastic/beats/libbeat/outputs/mode"
 	"github.com/elastic/beats/libbeat/outputs/mode/modeutil"
+	"github.com/elastic/beats/libbeat/outputs/transport"
 )
 
 type httpOutput struct {
@@ -71,8 +71,13 @@ func (out *httpOutput) init(cfg *common.Config) error {
 	var maxWaitRetry = time.Duration(60) * time.Second
 
 	loadBalance := config.LoadBalance
-	m, err := modeutil.NewConnectionMode(clients, !loadBalance,
-		maxAttempts, waitRetry, config.Timeout, maxWaitRetry)
+	m, err := modeutil.NewConnectionMode(clients, modeutil.Settings{
+		Failover:     !loadBalance,
+		MaxAttempts:  maxAttempts,
+		Timeout:      config.Timeout,
+		WaitRetry:    waitRetry,
+		MaxWaitRetry: maxWaitRetry,
+	})
 	if err != nil {
 		return err
 	}
@@ -83,7 +88,7 @@ func (out *httpOutput) init(cfg *common.Config) error {
 }
 
 func makeClientFactory(
-	tls *tls.Config,
+	tls *transport.TLSConfig,
 	config *httpConfig,
 	out *httpOutput,
 ) func(string) (mode.ProtocolClient, error) {
@@ -111,10 +116,16 @@ func makeClientFactory(
 			params = nil
 		}
 
-		return NewClient(
-			hostURL, proxyURL, tls,
-			config.Username, config.Password,
-			params, config.Timeout, config.CompressionLevel)
+		return NewClient(ClientSettings{
+			URL:              hostURL,
+			Proxy:            proxyURL,
+			TLS:              tls,
+			Username:         config.Username,
+			Password:         config.Password,
+			Parameters:       params,
+			Timeout:          config.Timeout,
+			CompressionLevel: config.CompressionLevel,
+		})
 	}
 }
 
