Support Elasticsearch HTTPS & Basic Auth

README.md

@@ -171,6 +171,13 @@ First, set the IP address and port where the agent can find the Elasticsearch in
     # Set the host and port where to find Elasticsearch.
     host = "192.168.1.42"
     port = 9200
+
+    # Optional protocol
+    # protocol = "https"
+
+    # Optional basic auth credentials
+    # username = "admin"
+    # password = "s3cr3t"
 ```
 
 Select the network interface from which to capture the traffic. Packetbeat supports capturing all messages sent or received by the server on which it is installed. For this, use "any" as the device: 

packetbeat.conf

@@ -13,6 +13,13 @@
 host = "localhost"
 port = 9200
 
+# Optional protocol
+# protocol = "https"
+
+# Optional basic auth credentials
+# username = "admin"
+# password = "s3cr3t"
+
 [interfaces]
 # Select on which network interfaces to sniff. You can use the "any"
 # keyword to sniff on all connected interfaces.

publish.go

@@ -15,10 +15,6 @@ import (
 type PublisherType struct {
     name string
 
-    url         string
-    mother_host string
-    mother_port string
-
     RefreshTopologyTimer <-chan time.Time
     TopologyMap          map[string]string
 }
@@ -31,8 +27,11 @@ type tomlAgent struct {
     Refresh_topology_freq int
 }
 type tomlMothership struct {
-    Host string
-    Port int
+    Host     string
+    Port     int
+    Protocol string
+    Username string
+    Password string
 }
 
 type Event struct {
@@ -82,10 +81,6 @@ func (publisher *PublisherType) GetServerName(ip string) string {
 }
 
 func (publisher *PublisherType) PublishHttpTransaction(t *HttpTransaction) error {
-    // Set the Elasticsearch Host to Connect to
-    api.Domain = publisher.mother_host
-    api.Port = publisher.mother_port
-
     index := fmt.Sprintf("packetbeat-%d.%02d.%02d", t.ts.Year(), t.ts.Month(), t.ts.Day())
 
     status := t.Http["response"].(bson.M)["phrase"].(string)
@@ -122,10 +117,6 @@ func (publisher *PublisherType) PublishHttpTransaction(t *HttpTransaction) error
 }
 
 func (publisher *PublisherType) PublishMysqlTransaction(t *MysqlTransaction) error {
-    // Set the Elasticsearch Host to Connect to
-    api.Domain = publisher.mother_host
-    api.Port = publisher.mother_port
-
     index := fmt.Sprintf("packetbeat-%d.%02d.%02d", t.ts.Year(), t.ts.Month(), t.ts.Day())
 
     status := t.Mysql["error_message"].(string)
@@ -150,10 +141,6 @@ func (publisher *PublisherType) PublishMysqlTransaction(t *MysqlTransaction) err
 }
 
 func (publisher *PublisherType) PublishRedisTransaction(t *RedisTransaction) error {
-    // Set the Elasticsearch Host to Connect to
-    api.Domain = publisher.mother_host
-    api.Port = publisher.mother_port
-
     index := fmt.Sprintf("packetbeat-%d.%02d.%02d", t.ts.Year(), t.ts.Month(), t.ts.Day())
 
     status := "OK"
@@ -182,10 +169,6 @@ func (publisher *PublisherType) UpdateTopologyPeriodically() {
 
 func (publisher *PublisherType) UpdateTopology() {
 
-    // Set the Elasticsearch Host to Connect to
-    api.Domain = publisher.mother_host
-    api.Port = publisher.mother_port
-
     DEBUG("publish", "Updating Topology")
 
     // get all agents IPs from Elasticsearch
@@ -213,10 +196,6 @@ func (publisher *PublisherType) UpdateTopology() {
 
 func (publisher *PublisherType) PublishTopology(params ...string) error {
 
-    // Set the Elasticsearch Host to Connect to
-    api.Domain = publisher.mother_host
-    api.Port = publisher.mother_port
-
     var localAddrs []string = params
 
     if len(params) == 0 {
@@ -285,11 +264,17 @@ func (publisher *PublisherType) PublishTopology(params ...string) error {
 func (publisher *PublisherType) Init() error {
     var err error
 
-    publisher.mother_host = _Config.Elasticsearch.Host
-    publisher.mother_port = fmt.Sprintf("%d", _Config.Elasticsearch.Port)
+    // Set the Elasticsearch Host to Connect to
+    api.Domain = _Config.Elasticsearch.Host
+    api.Port = fmt.Sprintf("%d", _Config.Elasticsearch.Port)
+    api.Username = _Config.Elasticsearch.Username
+    api.Password = _Config.Elasticsearch.Password
+
+    if _Config.Elasticsearch.Protocol != "" {
+        api.Protocol = _Config.Elasticsearch.Protocol
+    }
 
-    publisher.url = fmt.Sprintf("%s:%s", publisher.mother_host, publisher.mother_port)
-    INFO("Use %s as publisher", publisher.url)
+    INFO("Use %s://%s:%s as publisher", api.Protocol, api.Domain, api.Port)
 
     publisher.name = _Config.Agent.Name
     if len(publisher.name) == 0 {

publish_test.go

@@ -14,9 +14,9 @@ func TestTopology(t *testing.T) {
     api.Port = "9200"
 
     _, _ = core.Delete("packetbeat-topology", "server-ip", "", nil)
-    var publisher1 PublisherType = PublisherType{name: "proxy1", mother_host: api.Domain, mother_port: api.Port}
-    var publisher2 PublisherType = PublisherType{name: "proxy2", mother_host: api.Domain, mother_port: api.Port}
-    var publisher3 PublisherType = PublisherType{name: "proxy3", mother_host: api.Domain, mother_port: api.Port}
+    var publisher1 PublisherType = PublisherType{name: "proxy1"}
+    var publisher2 PublisherType = PublisherType{name: "proxy2"}
+    var publisher3 PublisherType = PublisherType{name: "proxy3"}
 
     publisher1.PublishTopology("10.1.0.4")
     publisher2.PublishTopology("10.1.0.9", "fe80::4e8d:79ff:fef2:de6a")
@@ -73,7 +73,7 @@ func TestGetServerName(t *testing.T) {
     api.Domain = "10.0.50.4"
     api.Port = "9200"
 
-    var publisher PublisherType = PublisherType{name: "proxy1", mother_host: api.Domain, mother_port: api.Port, RefreshTopologyTimer: time.Tick(1 * time.Second)}
+    var publisher PublisherType = PublisherType{name: "proxy1", RefreshTopologyTimer: time.Tick(1 * time.Second)}
 
     name := publisher.GetServerName("127.0.0.1")
     if name != "proxy1" {