geoip system tests

docker-compose.yml

@@ -37,7 +37,6 @@ services:
       - "./testing/docker/elasticsearch/roles.yml:/usr/share/elasticsearch/config/roles.yml"
       - "./testing/docker/elasticsearch/users:/usr/share/elasticsearch/config/users"
       - "./testing/docker/elasticsearch/users_roles:/usr/share/elasticsearch/config/users_roles"
-      - "./testing/docker/elasticsearch/ingest-geoip:/usr/share/elasticsearch/config/ingest-geoip"
     logging: *default-logging
 
   kibana:

systemtest/containers.go

@@ -20,7 +20,6 @@ package systemtest
 import (
 	"bytes"
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -52,6 +51,8 @@ const (
 
 var (
 	systemtestDir string
+	agentImageMu  sync.RWMutex
+	agentImages   = make(map[string]bool)
 )
 
 func initContainers() {
@@ -435,18 +436,6 @@ func (c *ElasticAgentContainer) Exec(ctx context.Context, cmd ...string) (stdout
 	return stdoutBuf.Bytes(), stderrBuf.Bytes(), nil
 }
 
-func matchFleetServerAPIStatusHealthy(r io.Reader) bool {
-	var status struct {
-		Name    string `json:"name"`
-		Version string `json:"version"`
-		Status  string `json:"status"`
-	}
-	if err := json.NewDecoder(r).Decode(&status); err != nil {
-		return false
-	}
-	return status.Status == "HEALTHY"
-}
-
 // BuildElasticAgentImage builds a Docker image from the published image with a locally built apm-server injected.
 func BuildElasticAgentImage(
 	ctx context.Context,
@@ -478,8 +467,3 @@ func BuildElasticAgentImage(
 	agentImages[arch] = true
 	return nil
 }
-
-var (
-	agentImageMu sync.RWMutex
-	agentImages  = make(map[string]bool)
-)

systemtest/main_test.go

@@ -18,9 +18,17 @@
 package systemtest
 
 import (
+	"encoding/json"
+	"fmt"
+	"io"
 	"log"
+	"net/http"
 	"os"
+	"strings"
 	"testing"
+	"time"
+
+	"github.com/elastic/apm-server/systemtest/apmservertest"
 )
 
 func TestMain(m *testing.M) {
@@ -30,9 +38,87 @@ func TestMain(m *testing.M) {
 		log.Fatalf("failed to start stack containers: %v", err)
 	}
 	initElasticSearch()
+	if err := GeoIpLazyDownload(); err != nil {
+		log.Fatalf("failed to download geoip database: %v", err)
+	}
 	initKibana()
 	initSettings()
 	initOTEL()
 	log.Println("INFO: running system tests...")
 	os.Exit(m.Run())
 }
+
+const requestBody = `{"metadata": { "service": {"name": "alice", "agent": {"version": "3.14.0", "name": "elastic-node"}}}}
+{"metricset": {"samples":{"a":{"value":3.2}}, "timestamp": 1496170422281000}}`
+
+func GeoIpLazyDownload() error {
+	srv := apmservertest.NewUnstartedServer()
+	if err := srv.Start(); err != nil {
+		return err
+	}
+
+	serverURL := srv.URL + "/intake/v2/events"
+	req, err := http.NewRequest("POST", serverURL, strings.NewReader(requestBody))
+	if err != nil {
+		return err
+	}
+
+	req.Header.Set("Content-Type", "application/x-ndjson")
+	req.Header.Set("X-Forwarded-For", "8.8.8.8")
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return err
+	}
+	if resp.StatusCode != http.StatusAccepted {
+		return fmt.Errorf("query failed with status code: %d", resp.StatusCode)
+	}
+
+	io.Copy(io.Discard, resp.Body)
+	resp.Body.Close()
+
+	if err := waitGeoIPDownload(); err != nil {
+		return err
+	}
+
+	// Clean all datastreams containing tags.
+	cleanupElasticsearch()
+
+	return nil
+}
+
+func waitGeoIPDownload() error {
+	timer := time.NewTimer(2 * time.Minute)
+	ticker := time.NewTicker(500 * time.Millisecond)
+	defer ticker.Stop()
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			resp, err := Elasticsearch.Ingest.GeoIPStats()
+			if err != nil {
+				return err
+			}
+			body, err := io.ReadAll(resp.Body)
+			resp.Body.Close()
+			if err != nil {
+				return err
+			}
+			var stats = struct {
+				Stats struct {
+					DatabasesCount int `json:"databases_count"`
+				} `json:"stats"`
+			}{}
+			if err := json.Unmarshal(body, &stats); err != nil {
+				return err
+			}
+			if stats.Stats.DatabasesCount == 3 {
+				log.Println("GeoIP database downloaded")
+				return nil
+			}
+		case <-timer.C:
+			return fmt.Errorf("download timeout exceeded")
+		}
+	}
+}