Add sleep to allow ES sufficient time for CCR

metricbeat/module/elasticsearch/elasticsearch_integration_test.go

@@ -28,6 +28,7 @@ import (
  "net/http"
  "os"
  "testing"
+ "time"
 
  "github.com/pkg/errors"
 
@@ -255,9 +256,48 @@ func createCCRStats(host string) error {
  return errors.Wrap(err, "error creating CCR follower index")
  }
 
+ // Give ES sufficient time to do the replication and produce stats
+ 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 {
@@ -364,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
+}