Only connect to Elasticsearch instances with the same version or newer

CHANGELOG.next.asciidoc

@@ -138,6 +138,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add FIPS configuration option for all AWS API calls. {pull}28899[28899]
 - Add `default_region` config to AWS common module. {pull}29415[29415]
 - Add support for latest k8s versions v1.23 and v1.22 {pull}29575[29575]
+- Only connect to Elasticsearch instances with the same version or newer. {pull}29683[29683]
 
 *Auditbeat*
 

auditbeat/auditbeat.reference.yml

@@ -460,6 +460,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # auditbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

filebeat/Jenkinsfile.yml

@@ -48,6 +48,7 @@ stages:
       mage: >-                 ## Run module integration tests under previous minor of ES to ensure ingest pipeline compatibility.
         STACK_ENVIRONMENT=prev-minor
         TESTING_FILEBEAT_SKIP_DIFF=1
+        TESTING_FILEBEAT_ALLOW_OLDER=1
         PYTEST_ADDOPTS='-k test_modules'
         mage pythonIntegTest
       withModule: true

filebeat/filebeat.reference.yml

@@ -1404,6 +1404,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # filebeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

filebeat/tests/system/test_modules.py

@@ -124,6 +124,9 @@ def run_on_file(self, module, fileset, test_file, cfgfile):
                 module=module, fileset=fileset, test_file=test_file),
             "-M", "*.*.input.close_eof=true",
         ]
+        # allow connecting older versions of Elasticsearch
+        if os.getenv("TESTING_FILEBEAT_ALLOW_OLDER"):
+            cmd.extend(["-E", "output.elasticsearch.allow_older_versions=true"])
 
         # Based on the convention that if a name contains -json the json format is needed. Currently used for LS.
         if "-json" in test_file:

heartbeat/heartbeat.reference.yml

@@ -606,6 +606,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # heartbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

libbeat/_meta/config/output-elasticsearch.reference.yml.tmpl

@@ -77,6 +77,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # {{.BeatName}} expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
 {{include "ssl.reference.yml.tmpl" . | indent 2 }}
   # Enable Kerberos support. Kerberos is automatically enabled if any Kerberos setting is set.
   #kerberos.enabled: true

libbeat/cmd/instance/beat.go

@@ -326,6 +326,8 @@ func (b *Beat) createBeater(bt beat.Creator) (beat.Beater, error) {
 	logSystemInfo(b.Info)
 	logp.Info("Setup Beat: %s; Version: %s", b.Info.Beat, b.Info.Version)
 
+	b.checkElasticsearchVersion()
+
 	err = b.registerESIndexManagement()
 	if err != nil {
 		return nil, err
@@ -855,6 +857,37 @@ func (b *Beat) loadDashboards(ctx context.Context, force bool) error {
 	return nil
 }
 
+// checkElasticsearchVersion registers a global callback to make sure ES instance we are connecting
+// to is at least on the same version as the Beat.
+// If the check is disabled or the output is not Elasticsearch, nothing happens.
+func (b *Beat) checkElasticsearchVersion() {
+	if b.Config.Output.Name() != "elasticsearch" || b.isConnectionToOlderVersionAllowed() {
+		return
+	}
+
+	elasticsearch.RegisterGlobalCallback(func(conn *eslegclient.Connection) error {
+		esVersion := conn.GetVersion()
+		beatVersion, err := common.NewVersion(b.Info.Version)
+		if err != nil {
+			return err
+		}
+		if esVersion.LessThan(beatVersion) {
+			return fmt.Errorf("%v ES=%s, Beat=%s.", elasticsearch.ErrTooOld, esVersion.String(), b.Info.Version)
+		}
+		return nil
+	})
+}
+
+func (b *Beat) isConnectionToOlderVersionAllowed() bool {
+	config := struct {
+		AllowOlder bool `config:"allow_older_versions"`
+	}{false}
+
+	b.Config.Output.Config().Unpack(&config)
+
+	return config.AllowOlder
+}
+
 // registerESIndexManagement registers the loading of the template and ILM
 // policy as a callback with the elasticsearch output. It is important the
 // registration happens before the publisher is created.

libbeat/outputs/elasticsearch/client.go

@@ -38,7 +38,11 @@ import (
 	"github.com/elastic/beats/v7/libbeat/testing"
 )
 
-var errPayloadTooLarge = errors.New("the bulk payload is too large for the server. Consider to adjust `http.max_content_length` parameter in Elasticsearch or `bulk_max_size` in the beat. The batch has been dropped")
+var (
+	errPayloadTooLarge = errors.New("the bulk payload is too large for the server. Consider to adjust `http.max_content_length` parameter in Elasticsearch or `bulk_max_size` in the beat. The batch has been dropped")
+
+	ErrTooOld = errors.New("Elasticsearch is too old. Please upgrade the instance. If you would like to connect to older instances set output.elasticsearch.allow_older_versions to true.")
+)
 
 // Client is an elasticsearch client.
 type Client struct {

libbeat/outputs/elasticsearch/config.go

@@ -43,6 +43,7 @@ type elasticsearchConfig struct {
 	MaxRetries         int                     `config:"max_retries"`
 	Backoff            Backoff                 `config:"backoff"`
 	NonIndexablePolicy *common.ConfigNamespace `config:"non_indexable_policy"`
+	AllowOlderVersion  bool                    `config:"allow_older_versions"`
 
 	Transport httpcommon.HTTPTransportSettings `config:",inline"`
 }

libbeat/outputs/elasticsearch/docs/elasticsearch.asciidoc

@@ -60,6 +60,9 @@ This output works with all compatible versions of Elasticsearch. See the
 https://www.elastic.co/support/matrix#matrix_compatibility[Elastic Support
 Matrix].
 
+For optimal experience, {beatname_uc} only connects to instances that are at least on the
+same version as the Beat. The check can be disabled by setting `output.elasticsearch.allow_older_versions`.
+
 ==== Configuration options
 
 You can specify the following options in the `elasticsearch` section of the +{beatname_lc}.yml+ config file:
@@ -668,6 +671,14 @@ Elasticsearch after a network error. The default is `60s`.
 
 The http request timeout in seconds for the Elasticsearch request. The default is 90.
 
+==== `allow_older_versions`
+
+By default, {beatname_uc} expects the Elasticsearch instance to be on the same or newer version to provide
+optimal experience. We suggest you connect to the same version to make sure all features {beatname_uc} is using are
+available in your Elasticsearch instance.
+
+You can disable the check for example during updating the Elastic Stack, so data collection can go on.
+
 ===== `ssl`
 
 Configuration options for SSL parameters like the certificate authority to use

libbeat/tests/system/config/mockbeat.yml.j2

@@ -42,6 +42,8 @@ output:
     {% for k, v in elasticsearch.items() -%}
     {{ k }}: {{ v }}
     {% endfor -%}
+    # older versions have to be allowed because mockbeat is on v9.9.9
+    allow_older_versions: true
   {%- endif %}
 
   # Redis as output

metricbeat/metricbeat.reference.yml

@@ -1323,6 +1323,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # metricbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

packetbeat/packetbeat.reference.yml

@@ -955,6 +955,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # packetbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

winlogbeat/winlogbeat.reference.yml

@@ -391,6 +391,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # winlogbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

x-pack/auditbeat/auditbeat.reference.yml

@@ -516,6 +516,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # auditbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

x-pack/filebeat/Jenkinsfile.yml

@@ -48,6 +48,7 @@ stages:
       mage: >-                 ## Run module integration tests under previous minor of ES to ensure ingest pipeline compatibility.
         STACK_ENVIRONMENT=prev-minor
         TESTING_FILEBEAT_SKIP_DIFF=1
+        TESTING_FILEBEAT_ALLOW_OLDER=1
         PYTEST_ADDOPTS='-k test_xpack_modules'
         mage pythonIntegTest
       withModule: true

x-pack/filebeat/filebeat.reference.yml

@@ -3634,6 +3634,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # filebeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

x-pack/functionbeat/functionbeat.reference.yml

@@ -638,6 +638,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # functionbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

x-pack/heartbeat/heartbeat.reference.yml

@@ -606,6 +606,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # heartbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

x-pack/metricbeat/metricbeat.reference.yml

@@ -1844,6 +1844,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # metricbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

x-pack/osquerybeat/osquerybeat.reference.yml

@@ -357,6 +357,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # osquerybeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

x-pack/packetbeat/packetbeat.reference.yml

@@ -955,6 +955,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # packetbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 

x-pack/winlogbeat/winlogbeat.reference.yml

@@ -434,6 +434,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # winlogbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true