Upgrade test Docker images to latest versions

metricbeat/module/elasticsearch/_meta/Dockerfile

@@ -1,2 +1,2 @@
-FROM docker.elastic.co/elasticsearch/elasticsearch:6.4.3
-HEALTHCHECK --interval=1s --retries=300 CMD curl -f http://localhost:9200
+FROM docker.elastic.co/elasticsearch/elasticsearch:6.5.1
+HEALTHCHECK --interval=1s --retries=300 CMD curl -f http://localhost:9200/_xpack/license

metricbeat/module/elasticsearch/ccr/_meta/test/test_follower_index.json

@@ -0,0 +1,4 @@
+{
+    "remote_cluster": "same",
+    "leader_index": "pied_piper"
+}

metricbeat/module/elasticsearch/ccr/_meta/test/test_leader_index.json

@@ -0,0 +1,5 @@
+{
+    "settings": {
+        "soft_deletes.enabled": true
+    }
+}

metricbeat/module/elasticsearch/ccr/_meta/test/test_remote_settings.json

@@ -0,0 +1,13 @@
+{
+    "transient": {
+        "cluster": {
+            "remote": {
+                "same": {
+                    "seeds": [
+                        "127.0.0.1:9300"
+                    ]
+                }
+            }
+        }
+    }
+}

metricbeat/module/elasticsearch/elasticsearch_integration_test.go

@@ -72,6 +72,9 @@ func TestFetch(t *testing.T) {
 	err = createMLJob(host)
 	assert.NoError(t, err)
 
+	err = createCCRStats(host)
+	assert.NoError(t, err)
+
 	for _, metricSet := range metricSets {
 		checkSkip(t, metricSet, host)
 		t.Run(metricSet, func(t *testing.T) {
@@ -187,36 +190,71 @@ func createMLJob(host string) error {
 		return err
 	}
 
-	client := &http.Client{}
-
 	jobURL := "/_xpack/ml/anomaly_detectors/total-requests"
 
 	if checkExists("http://" + host + jobURL) {
 		return nil
 	}
 
-	req, err := http.NewRequest("PUT", "http://"+host+jobURL, bytes.NewReader(mlJob))
+	body, resp, err := httpPutJSON(host, jobURL, mlJob)
+
+	if resp.StatusCode != 200 {
+		return fmt.Errorf("HTTP error loading ml job %d: %s, %s", resp.StatusCode, resp.Status, body)
+	}
+
+	return nil
+}
+
+func createCCRStats(host string) error {
+	err := setupCCRRemote(host)
 	if err != nil {
 		return err
 	}
-	req.Header.Add("Content-Type", "application/json")
 
-	resp, err := client.Do(req)
+	err = createCCRLeaderIndex(host)
 	if err != nil {
 		return err
 	}
-	defer resp.Body.Close()
 
-	body, err := ioutil.ReadAll(resp.Body)
+	err = createCCRFollowerIndex(host)
 	if err != nil {
 		return err
 	}
 
-	if resp.StatusCode != 200 {
-		return fmt.Errorf("HTTP error loading ml job %d: %s, %s", resp.StatusCode, resp.Status, body)
+	return nil
+}
+
+func setupCCRRemote(host string) error {
+	remoteSettings, err := ioutil.ReadFile("ccr/_meta/test/test_remote_settings.json")
+	if err != nil {
+		return err
 	}
 
-	return nil
+	settingsURL := "/_cluster/settings"
+	_, _, err = httpPutJSON(host, settingsURL, remoteSettings)
+	return err
+}
+
+func createCCRLeaderIndex(host string) error {
+	leaderIndex, err := ioutil.ReadFile("ccr/_meta/test/test_leader_index.json")
+	if err != nil {
+		return err
+	}
+
+	indexURL := "/pied_piper"
+	_, _, err = httpPutJSON(host, indexURL, leaderIndex)
+	return err
+}
+
+func createCCRFollowerIndex(host string) error {
+	followerIndex, err := ioutil.ReadFile("ccr/_meta/test/test_follower_index.json")
+	if err != nil {
+		return err
+	}
+
+	followURL := "/rats/_ccr/follow"
+	_, _, err = httpPutJSON(host, followURL, followerIndex)
+	return err
 }
 
 func checkExists(url string) bool {
@@ -277,3 +315,25 @@ func getElasticsearchVersion(elasticsearchHostPort string) (string, error) {
 	}
 	return version.(string), nil
 }
+
+func httpPutJSON(host, path string, body []byte) ([]byte, *http.Response, error) {
+	req, err := http.NewRequest("PUT", "http://"+host+path, bytes.NewReader(body))
+	if err != nil {
+		return nil, nil, err
+	}
+	req.Header.Add("Content-Type", "application/json")
+
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err = ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return body, resp, nil
+}

metricbeat/module/elasticsearch/test_elasticsearch.py

@@ -34,12 +34,15 @@ def test_metricsets(self, metricset):
         """
         elasticsearch metricset tests
         """
-        self.check_skip(metricset)
+        es = Elasticsearch(self.get_hosts())
+        self.check_skip(metricset, es)
 
+        self.start_trial(es)
         if metricset == "ml_job":
-            self.create_ml_job()
+            self.create_ml_job(es)
+        if metricset == "ccr":
+            self.create_ccr_stats(es)
 
-        es = Elasticsearch(self.get_hosts())
         es.indices.create(index='test-index', ignore=400)
         self.check_metricset("elasticsearch", metricset, self.get_hosts(), self.FIELDS +
                              ["service.name"], extras={"index_recovery.active_only": "false"})
@@ -48,16 +51,7 @@ def get_hosts(self):
         return [os.getenv('ES_HOST', 'localhost') + ':' +
                 os.getenv('ES_PORT', '9200')]
 
-    def create_ml_job(self):
-        es = Elasticsearch(self.get_hosts())
-
-        # Enable xpack trial
-        try:
-            es.transport.perform_request('POST', "/_xpack/license/start_trial?acknowledge=true")
-        except:
-            e = sys.exc_info()[0]
-            print "Trial already enabled. Error: {}".format(e)
-
+    def create_ml_job(self, es):
         # Check if an ml job already exists
         response = es.transport.perform_request('GET', "/_xpack/ml/anomaly_detectors/_all/")
         if response["count"] > 0:
@@ -72,21 +66,66 @@ def create_ml_job(self):
         path = "/_xpack/ml/anomaly_detectors/test"
         es.transport.perform_request('PUT', path, body=body)
 
-    def check_skip(self, metricset):
+    def create_ccr_stats(self, es):
+        self.setup_ccr_remote(es)
+        self.create_ccr_leader_index(es)
+        self.create_ccr_follower_index(es)
+
+    def setup_ccr_remote(self, es):
+        file = os.path.join(self.beat_path, "module", "elasticsearch", "ccr",
+                            "_meta", "test", "test_remote_settings.json")
+
+        body = {}
+        with open(file, 'r') as f:
+            body = json.load(f)
+
+        path = "/_cluster/settings"
+        es.transport.perform_request('PUT', path, body=body)
+
+    def create_ccr_leader_index(self, es):
+        file = os.path.join(self.beat_path, "module", "elasticsearch", "ccr", "_meta", "test", "test_leader_index.json")
+
+        body = {}
+        with open(file, 'r') as f:
+            body = json.load(f)
+
+        path = "/pied_piper"
+        es.transport.perform_request('PUT', path, body=body)
+
+    def create_ccr_follower_index(self, es):
+        file = os.path.join(self.beat_path, "module", "elasticsearch", "ccr",
+                            "_meta", "test", "test_follower_index.json")
+
+        body = {}
+        with open(file, 'r') as f:
+            body = json.load(f)
+
+        path = "/rats/_ccr/follow"
+        es.transport.perform_request('PUT', path, body=body)
+
+    def start_trial(self, es):
+        # Check if trial is already enabled
+        response = es.transport.perform_request('GET', "/_xpack/license")
+        if response["license"]["type"] == "trial":
+            return
+
+        # Enable xpack trial
+        try:
+            es.transport.perform_request('POST', "/_xpack/license/start_trial?acknowledge=true")
+        except:
+            e = sys.exc_info()[0]
+            print "Trial already enabled. Error: {}".format(e)
+
+    def check_skip(self, metricset, es):
         if metricset != "ccr":
             return
 
-        version = self.get_version()
+        version = self.get_version(es)
         if semver.compare(version, "6.5.0") == -1:
             # Skip CCR metricset system test for Elasticsearch versions < 6.5.0 as CCR Stats
             # API endpoint is not available
             raise SkipTest("elasticsearch/ccr metricset system test only valid with Elasticsearch versions >= 6.5.0")
 
-    def get_version(self):
-        host = self.get_hosts()[0]
-        res = urllib2.urlopen("http://" + host + "/").read()
-
-        body = json.loads(res)
-        version = body["version"]["number"]
-
-        return version
+    def get_version(self, es):
+        response = es.transport.perform_request('GET', "/")
+        return response["version"]["number"]

metricbeat/module/kibana/_meta/Dockerfile

@@ -1,2 +1,2 @@
-FROM docker.elastic.co/kibana/kibana:6.4.3
+FROM docker.elastic.co/kibana/kibana:6.5.1
 HEALTHCHECK --interval=1s --retries=300 CMD curl -f http://localhost:5601/api/status | grep '"disconnects"'