Remove the _ttl param from the Elasticsearch topology code (#1907)

Ttl is no longer supported by Elasticsearch, so we cannot use it.
The alternative is to use curator or similar to cleanup old values.

CHANGELOG.asciidoc

@@ -14,6 +14,8 @@ https://github.com/elastic/beats/compare/v5.0.0-alpha3...master[Check the HEAD d
 
 *Affecting all Beats*
 
+- The topology_expire option of the Elasticserach output was removed. {pull}1907[1907]
+
 *Metricbeat*
 
 *Packetbeat*

filebeat/filebeat.full.yml

@@ -236,7 +236,9 @@ filebeat.prospectors:
 
 # Expiration time (in seconds) of the IPs published by a shipper to the topology map.
 # All the IPs will be deleted afterwards. Note, that the value must be higher than
-# refresh_topology_freq. The default is 15 seconds.
+# refresh_topology_freq. This setting is used only by the Redis output. The other
+# outputs don't support expiring entries.
+# The default is 15 seconds.
 #topology_expire: 15
 
 # Internal queue size for single events in processing pipeline
@@ -333,10 +335,6 @@ output.elasticsearch:
   # false. This option makes sense only for Packetbeat.
   #save_topology: false
 
-  # The time to live in seconds for the topology information that is stored in
-  # Elasticsearch. The default is 15 seconds.
-  #topology_expire: 15
-
   # A template is used to set the mapping in Elasticsearch
   # By default template loading is enabled and the template is loaded.
   # These settings can be adjusted to load your own template or overwrite existing ones

libbeat/_meta/config.full.yml

@@ -33,7 +33,9 @@
 
 # Expiration time (in seconds) of the IPs published by a shipper to the topology map.
 # All the IPs will be deleted afterwards. Note, that the value must be higher than
-# refresh_topology_freq. The default is 15 seconds.
+# refresh_topology_freq. This setting is used only by the Redis output. The other
+# outputs don't support expiring entries.
+# The default is 15 seconds.
 #topology_expire: 15
 
 # Internal queue size for single events in processing pipeline
@@ -130,10 +132,6 @@ output.elasticsearch:
   # false. This option makes sense only for Packetbeat.
   #save_topology: false
 
-  # The time to live in seconds for the topology information that is stored in
-  # Elasticsearch. The default is 15 seconds.
-  #topology_expire: 15
-
   # A template is used to set the mapping in Elasticsearch
   # By default template loading is enabled and the template is loaded.
   # These settings can be adjusted to load your own template or overwrite existing ones

libbeat/docs/generalconfig.asciidoc

@@ -164,10 +164,14 @@ IP addresses to the topology map. The default is 10 seconds.
 
 ===== topology_expire
 
-The expiration time for the topology in seconds. This is
-useful in case a Beat stops publishing its IP addresses. The IP addresses
-are removed automatically from the topology map after expiration. The default
-is 15 seconds.
+The expiration time for the topology in seconds. This is useful in case a Beat
+stops publishing its IP addresses. The IP addresses are removed automatically
+from the topology map after expiration.
+
+This setting is used only by the Redis output. The other outputs don't support
+expiring entries.
+
+The default is 15 seconds.
 
 ===== queue_size
 

libbeat/docs/outputconfig.asciidoc

@@ -227,7 +227,7 @@ The maximum number of events to bulk in a single Elasticsearch bulk API index re
 
 If the Beat sends single events, the events are collected into batches. If the Beat publishes
 a large batch of events (larger than the value specified by `bulk_max_size`), the batch is
-split. 
+split.
 
 Specifying a larger batch size can improve performance by lowering the overhead of sending events. 
 However big batch sizes can also increase processing times, which might result in
@@ -257,11 +257,6 @@ false.
 
 This option is relevant for Packetbeat only.
 
-===== topology_expire
-
-The time to live in seconds for the topology information that is stored in
-Elasticsearch. The default is 15 seconds.
-
 ===== tls
 
 Configuration options for TLS parameters like the certificate authority to use

libbeat/outputs/elasticsearch/output.go

@@ -103,29 +103,6 @@ func (out *elasticsearchOutput) init(
 		return err
 	}
 
-	if config.SaveTopology {
-		err := out.EnableTTL()
-		if err != nil {
-			logp.Err("Fail to set _ttl mapping: %s", err)
-			// keep trying in the background
-			go func() {
-				for {
-					err := out.EnableTTL()
-					if err == nil {
-						break
-					}
-					logp.Err("Fail to set _ttl mapping: %s", err)
-					time.Sleep(5 * time.Second)
-				}
-			}()
-		}
-	}
-
-	out.TopologyExpire = 15000
-	if topologyExpire != 0 {
-		out.TopologyExpire = topologyExpire * 1000 // millisec
-	}
-
 	out.mode = m
 	out.index = config.Index
 

libbeat/outputs/elasticsearch/output_test.go

@@ -45,8 +45,6 @@ func createElasticsearchConnection(flushInterval int, bulkSize int) elasticsearc
 
 func TestTopologyInES(t *testing.T) {
 
-	t.Skip("This tests is skipped because it currently fails with 5.0.0-alpha4. Probably because _ttl was removed")
-
 	if testing.Verbose() {
 		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"topology", "output_elasticsearch"})
 	}
@@ -310,25 +308,3 @@ func TestBulkEvents(t *testing.T) {
 	output = createElasticsearchConnection(50, 5)
 	testBulkWithParams(t, output)
 }
-
-func TestEnableTTL(t *testing.T) {
-
-	t.Skip("This tests is skipped as ttl is not compatible with 5.0.0-alpha4 as _ttl was removed")
-	if testing.Verbose() {
-		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"topology", "output_elasticsearch", "elasticsearch"})
-	}
-
-	output := createElasticsearchConnection(0, 0)
-	output.randomClient().Delete(".packetbeat-topology", "", "", nil)
-
-	err := output.EnableTTL()
-	if err != nil {
-		t.Errorf("Fail to enable TTL: %s", err)
-	}
-
-	// should succeed also when index already exists
-	err = output.EnableTTL()
-	if err != nil {
-		t.Errorf("Fail to enable TTL: %s", err)
-	}
-}

libbeat/outputs/elasticsearch/topology.go

@@ -2,7 +2,6 @@ package elasticsearch
 
 import (
 	"encoding/json"
-	"fmt"
 	"math/rand"
 	"strconv"
 	"strings"
@@ -15,9 +14,7 @@ import (
 type topology struct {
 	clients []mode.ProtocolClient
 
-	TopologyExpire int
-	TopologyMap    atomic.Value // Value holds a map[string][string]
-	ttlEnabled     bool
+	TopologyMap atomic.Value // Value holds a map[string][string]
 }
 
 type publishedTopology struct {
@@ -36,34 +33,6 @@ func (t *topology) randomClient() *Client {
 	}
 }
 
-// Enable using ttl as paramters in a server-ip doc type
-func (t *topology) EnableTTL() error {
-	client := t.randomClient()
-	if client == nil {
-		return ErrNotConnected
-	}
-
-	setting := map[string]interface{}{
-		"server-ip": map[string]interface{}{
-			"_ttl": map[string]string{"enabled": "true", "default": "15s"},
-		},
-	}
-
-	// make sure the .packetbeat-topology index exists
-	// Ignore error here, as CreateIndex will error (400 Bad Request) if index
-	// already exists. If index could not be created, next api call to index will
-	// fail anyway.
-	index := ".packetbeat-topology"
-	_, _, _ = client.CreateIndex(index, nil)
-	_, _, err := client.Index(index, "server-ip", "_mapping", nil, setting)
-	if err != nil {
-		return err
-	}
-
-	t.ttlEnabled = true
-	return nil
-}
-
 // Get the name of a shipper by its IP address from the local topology map
 func (t *topology) GetNameByIP(ip string) string {
 	topologyMap, ok := t.TopologyMap.Load().(map[string]string)
@@ -78,20 +47,14 @@ func (t *topology) GetNameByIP(ip string) string {
 
 // Each shipper publishes a list of IPs together with its name to Elasticsearch
 func (t *topology) PublishIPs(name string, localAddrs []string) error {
-	if !t.ttlEnabled {
-		debugf("Not publishing IPs because TTL was not yet confirmed to be enabled")
-		return nil
-	}
-
 	client := t.randomClient()
 	if client == nil {
 		return ErrNotConnected
 	}
 
-	debugf("Publish IPs %s with expiration time %d", localAddrs, t.TopologyExpire)
+	debugf("Publish IPs: %s", localAddrs)
 
 	params := map[string]string{
-		"ttl":     fmt.Sprintf("%dms", t.TopologyExpire),
 		"refresh": "true",
 	}
 	_, _, err := client.Index(

metricbeat/metricbeat.full.yml

@@ -178,7 +178,9 @@ metricbeat.modules:
 
 # Expiration time (in seconds) of the IPs published by a shipper to the topology map.
 # All the IPs will be deleted afterwards. Note, that the value must be higher than
-# refresh_topology_freq. The default is 15 seconds.
+# refresh_topology_freq. This setting is used only by the Redis output. The other
+# outputs don't support expiring entries.
+# The default is 15 seconds.
 #topology_expire: 15
 
 # Internal queue size for single events in processing pipeline
@@ -275,10 +277,6 @@ output.elasticsearch:
   # false. This option makes sense only for Packetbeat.
   #save_topology: false
 
-  # The time to live in seconds for the topology information that is stored in
-  # Elasticsearch. The default is 15 seconds.
-  #topology_expire: 15
-
   # A template is used to set the mapping in Elasticsearch
   # By default template loading is enabled and the template is loaded.
   # These settings can be adjusted to load your own template or overwrite existing ones

packetbeat/packetbeat.full.yml

@@ -417,7 +417,9 @@ packetbeat.protocols.nfs:
 
 # Expiration time (in seconds) of the IPs published by a shipper to the topology map.
 # All the IPs will be deleted afterwards. Note, that the value must be higher than
-# refresh_topology_freq. The default is 15 seconds.
+# refresh_topology_freq. This setting is used only by the Redis output. The other
+# outputs don't support expiring entries.
+# The default is 15 seconds.
 #topology_expire: 15
 
 # Internal queue size for single events in processing pipeline
@@ -514,10 +516,6 @@ output.elasticsearch:
   # false. This option makes sense only for Packetbeat.
   #save_topology: false
 
-  # The time to live in seconds for the topology information that is stored in
-  # Elasticsearch. The default is 15 seconds.
-  #topology_expire: 15
-
   # A template is used to set the mapping in Elasticsearch
   # By default template loading is enabled and the template is loaded.
   # These settings can be adjusted to load your own template or overwrite existing ones

winlogbeat/winlogbeat.full.yml

@@ -68,7 +68,9 @@ winlogbeat.event_logs:
 
 # Expiration time (in seconds) of the IPs published by a shipper to the topology map.
 # All the IPs will be deleted afterwards. Note, that the value must be higher than
-# refresh_topology_freq. The default is 15 seconds.
+# refresh_topology_freq. This setting is used only by the Redis output. The other
+# outputs don't support expiring entries.
+# The default is 15 seconds.
 #topology_expire: 15
 
 # Internal queue size for single events in processing pipeline
@@ -165,10 +167,6 @@ output.elasticsearch:
   # false. This option makes sense only for Packetbeat.
   #save_topology: false
 
-  # The time to live in seconds for the topology information that is stored in
-  # Elasticsearch. The default is 15 seconds.
-  #topology_expire: 15
-
   # A template is used to set the mapping in Elasticsearch
   # By default template loading is enabled and the template is loaded.
   # These settings can be adjusted to load your own template or overwrite existing ones