Implement waitFor pattern instead of straight up sleep

metricbeat/module/elasticsearch/elasticsearch_integration_test.go

@@ -257,11 +257,47 @@ func createCCRStats(host string) error {
 	}
 
 	// Give ES sufficient time to do the replication and produce stats
-	time.Sleep(300 * time.Millisecond)
+	checkCCRStats := func() (bool, error) {
+		return checkCCRStatsExists(host)
+	}
+
+	exists, err := waitForSuccess(checkCCRStats, 200, 5)
+	if err != nil {
+		return errors.Wrap(err, "error checking if CCR stats exist")
+	}
+
+	if !exists {
+		return fmt.Errorf("expected to find CCR stats but not found")
+	}
 
 	return nil
 }
 
+func checkCCRStatsExists(host string) (bool, error) {
+	resp, err := http.Get("http://" + host + "/_ccr/stats")
+	if err != nil {
+		return false, err
+	}
+	defer resp.Body.Close()
+
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return false, err
+	}
+
+	var data struct {
+		FollowStats struct {
+			Indices []map[string]interface{} `json:"indices"`
+		} `json:"follow_stats"`
+	}
+	err = json.Unmarshal(body, &data)
+	if err != nil {
+		return false, err
+	}
+
+	return len(data.FollowStats.Indices) > 0, nil
+}
+
 func setupCCRRemote(host string) error {
 	remoteSettings, err := ioutil.ReadFile("ccr/_meta/test/test_remote_settings.json")
 	if err != nil {
@@ -368,3 +404,23 @@ func httpPutJSON(host, path string, body []byte) ([]byte, *http.Response, error)
 
 	return body, resp, nil
 }
+
+type checkSuccessFunction func() (bool, error)
+
+func waitForSuccess(f checkSuccessFunction, retryIntervalMs time.Duration, numAttempts int) (bool, error) {
+	for numAttempts > 0 {
+		success, err := f()
+		if err != nil {
+			return false, err
+		}
+
+		if success {
+			return success, nil
+		}
+
+		time.Sleep(retryIntervalMs * time.Millisecond)
+		numAttempts--
+	}
+
+	return false, nil
+}