Add TLS configuration to KafkaInput.

Author: Matt Moyer

docs/source/config/inputs/kafka.rst

@@ -77,6 +77,18 @@ Config:
     client code consumes events, greatly improving throughput. The default is
     16.
 
+.. versionadded:: 0.11
+
+- use_tls (bool, optional):
+    Specifies whether or not SSL/TLS encryption should be used for the TCP
+    connections. Defaults to false.
+
+- tls (TlsConfig, optional):
+    A sub-section that specifies the settings to be used for any SSL/TLS
+    encryption. This will only have any impact if ``use_tls`` is set to true.
+    See :ref:`tls`.
+
+
 Example 1: Read Fxa messages from partition 0.
 
 .. code-block:: ini

plugins/kafka/kafka_input.go

@@ -10,6 +10,7 @@
 # Contributor(s):
 #   Mike Trinkala ([email protected])
 #   Rob Miller ([email protected])
+#   Matt Moyer ([email protected])
 #
 # ***** END LICENSE BLOCK *****/
 
@@ -27,6 +28,7 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/mozilla-services/heka/message"
 	"github.com/mozilla-services/heka/pipeline"
+	"github.com/mozilla-services/heka/plugins/tcp"
 )
 
 type KafkaInputConfig struct {
@@ -39,6 +41,10 @@ type KafkaInputConfig struct {
 	WaitForElection            uint32 `toml:"wait_for_election"`
 	BackgroundRefreshFrequency uint32 `toml:"background_refresh_frequency"`
 
+	// TLS Config
+	UseTls bool `toml:"use_tls"`
+	Tls    tcp.TlsConfig
+
 	// Broker Config
 	MaxOpenRequests int    `toml:"max_open_reqests"`
 	DialTimeout     uint32 `toml:"dial_timeout"`
@@ -146,6 +152,13 @@ func (k *KafkaInput) Init(config interface{}) (err error) {
 	k.saramaConfig.Metadata.Retry.Backoff = time.Duration(k.config.WaitForElection) * time.Millisecond
 	k.saramaConfig.Metadata.RefreshFrequency = time.Duration(k.config.BackgroundRefreshFrequency) * time.Millisecond
 
+	k.saramaConfig.Net.TLS.Enable = k.config.UseTls
+	if k.config.UseTls {
+		if k.saramaConfig.Net.TLS.Config, err = tcp.CreateGoTlsConfig(&k.config.Tls); err != nil {
+			return fmt.Errorf("TLS init error: %s", err)
+		}
+	}
+
 	k.saramaConfig.Net.MaxOpenRequests = k.config.MaxOpenRequests
 	k.saramaConfig.Net.DialTimeout = time.Duration(k.config.DialTimeout) * time.Millisecond
 	k.saramaConfig.Net.ReadTimeout = time.Duration(k.config.ReadTimeout) * time.Millisecond