add lazy downloader to system tests that require geoip db

Author: rubvs

systemtest/agentconfig_test.go

@@ -39,6 +39,7 @@ import (
 
 func TestAgentConfig(t *testing.T) {
 	systemtest.CleanupElasticsearch(t)
+	systemtest.GeoIpLazyDownload(t)
 
 	serviceName := "systemtest_service"
 	serviceEnvironment := "testing"

systemtest/aggregation_test.go

@@ -39,6 +39,8 @@ import (
 
 func TestTransactionAggregation(t *testing.T) {
 	systemtest.CleanupElasticsearch(t)
+	systemtest.GeoIpLazyDownload(t)
+
 	srv := apmservertest.NewUnstartedServerTB(t)
 	require.NoError(t, srv.Start())
 
@@ -118,6 +120,8 @@ func TestTransactionAggregation(t *testing.T) {
 
 func TestTransactionAggregationShutdown(t *testing.T) {
 	systemtest.CleanupElasticsearch(t)
+	systemtest.GeoIpLazyDownload(t)
+
 	srv := apmservertest.NewServerTB(t)
 
 	// Send a transaction to the server to be aggregated.
@@ -146,6 +150,8 @@ func TestTransactionAggregationShutdown(t *testing.T) {
 func TestServiceDestinationAggregation(t *testing.T) {
 	t.Setenv("ELASTIC_APM_GLOBAL_LABELS", "department_name=apm,organization=observability,company=elastic")
 	systemtest.CleanupElasticsearch(t)
+	systemtest.GeoIpLazyDownload(t)
+
 	srv := apmservertest.NewServerTB(t)
 
 	// Send spans to the server to be aggregated.

systemtest/geoip.go

@@ -27,6 +27,8 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/apm-server/systemtest/apmservertest"
 )
 
 var stats struct {
@@ -37,66 +39,59 @@ type GeoIPStats struct {
 	DatabasesCount int `json:"databases_count"`
 }
 
-const requestBody = `{"metadata":{"service":{"name":"rum-js-test","agent":{"name":"rum-js","version":"5.5.0"}}}}
-{"transaction":{"trace_id":"611f4fa950f04631aaaaaaaaaaaaaaaa","id":"611f4fa950f04631","type":"page-load","duration":643,"span_count":{"started":0}}}
-{"metricset":{"samples":{"transaction.breakdown.count":{"value":12},"transaction.duration.sum.us":{"value":12},"transaction.duration.count":{"value":2},"transaction.self_time.sum.us":{"value":10},"transaction.self_time.count":{"value":2},"span.self_time.count":{"value":1},"span.self_time.sum.us":{"value":633.288}},"transaction":{"type":"request","name":"GET /"},"span":{"type":"external","subtype":"http"},"timestamp": 1496170422281000}}
-`
+const requestBody = `{"metadata": { "service": {"name": "alice", "agent": {"version": "3.14.0", "name": "elastic-node"}}}}
+{"metricset": {"samples":{"a":{"value":3.2}}, "timestamp": 1496170422281000}}`
 
-func GeoIpLazyDownload(t *testing.T, url string) {
-	t.Helper()
+func GeoIpLazyDownload(t testing.TB) {
+	srv := apmservertest.NewUnstartedServerTB(t)
+	require.NoError(t, srv.Start())
 
-	req, err := http.NewRequest("POST", url, strings.NewReader(requestBody))
+	serverURL := srv.URL + "/intake/v2/events"
+	req, err := http.NewRequest("POST", serverURL, strings.NewReader(requestBody))
 	require.NoError(t, 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)
 	require.NoError(t, err)
 	require.Equal(t, http.StatusAccepted, resp.StatusCode)
-
 	io.Copy(io.Discard, resp.Body)
 	resp.Body.Close()
 
-	err = waitGeoIPDownload(t)
-	require.NoError(t, err)
+	require.NoError(t, waitGeoIPDownload(t))
 
 	// Clean all datastreams containing tags.
 	CleanupElasticsearch(t)
 }
 
-func waitGeoIPDownload(t *testing.T) error {
+func waitGeoIPDownload(t testing.TB) error {
 	t.Helper()
-
-	timer := time.NewTimer(30 * time.Second)
+	timer := time.NewTimer(2 * time.Minute)
 	ticker := time.NewTicker(500 * time.Millisecond)
 	defer ticker.Stop()
+	defer timer.Stop()
 
-	first := true
 	for {
 		select {
 		case <-ticker.C:
 			resp, err := Elasticsearch.Ingest.GeoIPStats()
-			require.NoError(t, err)
-
+			if err != nil {
+				return err
+			}
 			body, err := io.ReadAll(resp.Body)
-			require.NoError(t, err)
-
-			err = json.Unmarshal([]byte(body), &stats)
-			require.NoError(t, err)
-
-			// [GeoLite2-ASN.mmdb, GeoLite2-City.mmdb, GeoLite2-Country.mmdb]
+			resp.Body.Close()
+			if err != nil {
+				return err
+			}
+			if err := json.Unmarshal(body, &stats); err != nil {
+				return err
+			}
 			if stats.DatabasesCount == 3 {
-				t.Log("GeoIp database downloaded")
+				t.Log("GeoIP database downloaded")
 				return nil
 			}
 		case <-timer.C:
 			return fmt.Errorf("download timeout exceeded")
 		}
-
-		if first {
-			t.Log("waiting for GeoIP database to download")
-			first = false
-		}
 	}
 }

systemtest/rum_test.go

@@ -40,6 +40,7 @@ import (
 
 func TestRUMXForwardedFor(t *testing.T) {
 	systemtest.CleanupElasticsearch(t)
+	systemtest.GeoIpLazyDownload(t)
 
 	srv := apmservertest.NewUnstartedServerTB(t)
 	srv.Config.RUM = &apmservertest.RUMConfig{Enabled: true}
@@ -50,8 +51,6 @@ func TestRUMXForwardedFor(t *testing.T) {
 	require.NoError(t, err)
 	serverURL.Path = "/intake/v2/rum/events"
 
-	systemtest.GeoIpLazyDownload(t, serverURL.String())
-
 	const body = `{"metadata":{"service":{"name":"rum-js-test","agent":{"name":"rum-js","version":"5.5.0"}}}}
 {"transaction":{"trace_id":"611f4fa950f04631aaaaaaaaaaaaaaaa","id":"611f4fa950f04631","type":"page-load","duration":643,"span_count":{"started":0}}}
 {"metricset":{"samples":{"transaction.breakdown.count":{"value":12},"transaction.duration.sum.us":{"value":12},"transaction.duration.count":{"value":2},"transaction.self_time.sum.us":{"value":10},"transaction.self_time.count":{"value":2},"span.self_time.count":{"value":1},"span.self_time.sum.us":{"value":633.288}},"transaction":{"type":"request","name":"GET /"},"span":{"type":"external","subtype":"http"},"timestamp": 1496170422281000}}
@@ -86,10 +85,9 @@ func TestRUMXForwardedFor(t *testing.T) {
 			Field:  "processor.event",
 			Values: []any{"transaction", "metric"},
 		},
-		//		espoll.WithTimeout(30*time.Second),
 	)
 
-	// Also checks for the absence of `tags` fields.
+	// Includes checking for absence of `tags` field.
 	approvaltest.ApproveFields(
 		t, t.Name(), result.Hits.Hits,
 		// RUM timestamps are set by the server based on the time the payload is received.