systemtest: use Fleet Server (#5085)

* systemtest: use Fleet Server

* systemtest: fix bulk unenrolment

Author: axw

systemtest/apmservertest/config.go

@@ -453,7 +453,7 @@ func defaultOutputConfig() OutputConfig {
 			Enabled: true,
 			Hosts: []string{net.JoinHostPort(
 				getenvDefault("ES_HOST", defaultElasticsearchHost),
-				getenvDefault("ES_PORT", defaultElasticsearchPort),
+				ElasticsearchPort(),
 			)},
 			Username: getenvDefault("ES_USER", defaultElasticsearchUser),
 			Password: getenvDefault("ES_PASS", defaultElasticsearchPass),
@@ -470,6 +470,13 @@ func KibanaPort() string {
 	return getenvDefault("KIBANA_PORT", defaultKibanaPort)
 }
 
+// ElasticsearchPort returns the Elasticsearch REST API port,
+// configured using ES_PORT, or otherwise returning the default
+// of 9200.
+func ElasticsearchPort() string {
+	return getenvDefault("ES_PORT", defaultElasticsearchPort)
+}
+
 func getenvDefault(k, defaultv string) string {
 	v := os.Getenv(k)
 	if v == "" {

systemtest/containers.go

@@ -43,7 +43,11 @@ import (
 	"github.com/elastic/go-elasticsearch/v7"
 )
 
-const startContainersTimeout = 5 * time.Minute
+const (
+	startContainersTimeout = 5 * time.Minute
+
+	defaultFleetServerPort = "8220"
+)
 
 // StartStackContainers starts Docker containers for Elasticsearch and Kibana.
 //
@@ -91,9 +95,7 @@ func NewUnstartedElasticsearchContainer() (*ElasticsearchContainer, error) {
 		Image:      container.Image,
 		AutoRemove: true,
 	}
-	waitFor := wait.ForHTTP("/")
-	waitFor.Port = "9200/tcp"
-	req.WaitingFor = waitFor
+	req.WaitingFor = wait.ForHTTP("/").WithPort("9200/tcp")
 
 	for port := range containerDetails.Config.ExposedPorts {
 		req.ExposedPorts = append(req.ExposedPorts, string(port))
@@ -252,30 +254,30 @@ func NewUnstartedElasticAgentContainer() (*ElasticAgentContainer, error) {
 	}
 	defer docker.Close()
 
-	kibanaContainer, err := stackContainerInfo(context.Background(), docker, "kibana")
+	esContainer, err := stackContainerInfo(context.Background(), docker, "elasticsearch")
 	if err != nil {
 		return nil, err
 	}
-	kibanaContainerDetails, err := docker.ContainerInspect(context.Background(), kibanaContainer.ID)
+	esContainerDetails, err := docker.ContainerInspect(context.Background(), esContainer.ID)
 	if err != nil {
 		return nil, err
 	}
 
-	var kibanaIPAddress string
+	var esIPAddress string
 	var networks []string
-	for network, settings := range kibanaContainerDetails.NetworkSettings.Networks {
+	for network, settings := range esContainerDetails.NetworkSettings.Networks {
 		networks = append(networks, network)
-		if kibanaIPAddress == "" && settings.IPAddress != "" {
-			kibanaIPAddress = settings.IPAddress
+		if esIPAddress == "" && settings.IPAddress != "" {
+			esIPAddress = settings.IPAddress
 		}
 	}
-	kibanaURL := &url.URL{
+	esURL := &url.URL{
 		Scheme: "http",
-		Host:   net.JoinHostPort(kibanaIPAddress, apmservertest.KibanaPort()),
+		Host:   net.JoinHostPort(esIPAddress, apmservertest.ElasticsearchPort()),
 	}
 
-	// Use the same stack version as used for Kibana.
-	agentImageVersion := kibanaContainer.Image[strings.LastIndex(kibanaContainer.Image, ":")+1:]
+	// Use the same stack version as used for Elasticsearch.
+	agentImageVersion := esContainer.Image[strings.LastIndex(esContainer.Image, ":")+1:]
 	agentImage := "docker.elastic.co/beats/elastic-agent:" + agentImageVersion
 	if err := pullDockerImage(context.Background(), docker, agentImage); err != nil {
 		return nil, err
@@ -293,11 +295,6 @@ func NewUnstartedElasticAgentContainer() (*ElasticAgentContainer, error) {
 		AutoRemove: true,
 		Networks:   networks,
 		Env: map[string]string{
-			"KIBANA_HOST": kibanaURL.String(),
-
-			// TODO(axw) remove once https://github.com/elastic/elastic-agent-client/issues/20 is fixed
-			"GODEBUG": "x509ignoreCN=0",
-
 			// NOTE(axw) because we bind-mount the apm-server artifacts in, they end up owned by the
 			// current user rather than root. Disable Beats's strict permission checks to avoid resulting
 			// complaints, as they're irrelevant to these system tests.
@@ -307,15 +304,17 @@ func NewUnstartedElasticAgentContainer() (*ElasticAgentContainer, error) {
 	return &ElasticAgentContainer{
 		request:          req,
 		installDir:       agentInstallDir,
+		elasticsearchURL: esURL.String(),
 		StackVersion:     agentImageVersion,
 		BindMountInstall: make(map[string]string),
 	}, nil
 }
 
 // ElasticAgentContainer represents an ephemeral Elastic Agent container.
 type ElasticAgentContainer struct {
-	container testcontainers.Container
-	request   testcontainers.ContainerRequest
+	container        testcontainers.Container
+	request          testcontainers.ContainerRequest
+	elasticsearchURL string // used by Fleet Server only
 
 	// installDir holds the location of the "install" directory inside
 	// the Elastic Agent container.
@@ -325,6 +324,10 @@ type ElasticAgentContainer struct {
 	// can be bind-mounted.
 	installDir string
 
+	// FleetServer controls whether this Elastic Agent bootstraps
+	// Fleet Server.
+	FleetServer bool
+
 	// StackVersion holds the stack version of the container image,
 	// e.g. 8.0.0-SNAPSHOT.
 	StackVersion string
@@ -335,6 +338,12 @@ type ElasticAgentContainer struct {
 	// WaitingFor holds an optional wait strategy.
 	WaitingFor wait.Strategy
 
+	// FleetServerURL holds the Fleet Server URL to enroll into.
+	//
+	// This will be populated by Start if FleetServer is true, using
+	// one of the container's network aliases.
+	FleetServerURL string
+
 	// Addrs holds the "host:port" address for each exposed port.
 	// This will be populated by Start.
 	Addrs []string
@@ -361,14 +370,27 @@ func (c *ElasticAgentContainer) Start() error {
 	defer cancel()
 
 	// Update request from user-definable fields.
-	if c.FleetEnrollmentToken != "" {
+	if c.FleetServer {
+		c.request.Env["FLEET_SERVER_ENABLE"] = "1"
+		c.request.Env["FLEET_SERVER_HOST"] = "0.0.0.0"
+		c.request.Env["FLEET_SERVER_PORT"] = defaultFleetServerPort
+		c.request.Env["FLEET_SERVER_INSECURE_HTTP"] = "1" // expose Fleet Server over HTTP
+		c.request.Env["FLEET_SERVER_ELASTICSEARCH_HOST"] = c.elasticsearchURL
+		c.request.Env["FLEET_SERVER_ELASTICSEARCH_USERNAME"] = adminElasticsearchUser
+		c.request.Env["FLEET_SERVER_ELASTICSEARCH_PASSWORD"] = adminElasticsearchPass
+
+		c.request.WaitingFor = wait.ForHTTP("/api/status").WithPort(defaultFleetServerPort + "/tcp")
+		c.request.ExposedPorts = []string{defaultFleetServerPort}
+	} else if c.FleetEnrollmentToken != "" {
 		c.request.Env["FLEET_ENROLL"] = "1"
 		c.request.Env["FLEET_ENROLLMENT_TOKEN"] = c.FleetEnrollmentToken
 		c.request.Env["FLEET_INSECURE"] = "1"
-		c.request.Env["FLEET_URL"] = "http://kibana:5601"
+		c.request.Env["FLEET_URL"] = c.FleetServerURL
+	}
+	c.request.ExposedPorts = append(c.request.ExposedPorts, c.ExposedPorts...)
+	if c.WaitingFor != nil {
+		c.request.WaitingFor = c.WaitingFor
 	}
-	c.request.ExposedPorts = c.ExposedPorts
-	c.request.WaitingFor = c.WaitingFor
 	c.request.BindMounts = map[string]string{}
 	for source, target := range c.BindMountInstall {
 		c.request.BindMounts[source] = path.Join(c.installDir, target)
@@ -390,15 +412,33 @@ func (c *ElasticAgentContainer) Start() error {
 		return err
 	}
 	if len(ports) > 0 {
-		ip, err := container.Host(ctx)
+		hostIP, err := container.Host(ctx)
 		if err != nil {
 			return err
 		}
 		for _, portbindings := range ports {
 			for _, pb := range portbindings {
-				c.Addrs = append(c.Addrs, net.JoinHostPort(ip, pb.HostPort))
+				c.Addrs = append(c.Addrs, net.JoinHostPort(hostIP, pb.HostPort))
+			}
+		}
+	}
+	if c.FleetServer {
+		networkAliases, err := container.NetworkAliases(ctx)
+		if err != nil {
+			return err
+		}
+		var networkAlias string
+		for _, networkAliases := range networkAliases {
+			if len(networkAliases) > 0 {
+				networkAlias = networkAliases[0]
+				break
 			}
 		}
+		if networkAlias == "" {
+			return errors.New("no network alias found")
+		}
+		fleetServerURL := &url.URL{Scheme: "http", Host: net.JoinHostPort(networkAlias, defaultFleetServerPort)}
+		c.FleetServerURL = fleetServerURL.String()
 	}
 
 	c.container = container

systemtest/fleet_test.go

@@ -28,6 +28,7 @@ import (
 	"path/filepath"
 	"runtime"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -62,9 +63,19 @@ func TestFleetIntegration(t *testing.T) {
 	err = fleet.CreatePackagePolicy(packagePolicy)
 	require.NoError(t, err)
 
+	// Bootstrap fleet-server.
+	fleetServer, err := systemtest.NewUnstartedElasticAgentContainer()
+	require.NoError(t, err)
+	fleetServer.FleetServer = true
+	defer fleetServer.Close()
+	err = fleetServer.Start()
+	require.NoError(t, err)
+
+	// Enroll another elastic-agent to run the APM integration.
 	agent, err := systemtest.NewUnstartedElasticAgentContainer()
 	require.NoError(t, err)
 	agent.FleetEnrollmentToken = enrollmentAPIKey.APIKey
+	agent.FleetServerURL = fleetServer.FleetServerURL
 	defer agent.Close()
 
 	defer func() {
@@ -101,9 +112,7 @@ func TestFleetIntegration(t *testing.T) {
 	// Start elastic-agent with port 8200 exposed, and wait for the server to service
 	// healthcheck requests to port 8200.
 	agent.ExposedPorts = []string{"8200"}
-	waitFor := wait.ForHTTP("/")
-	waitFor.Port = "8200/tcp"
-	agent.WaitingFor = waitFor
+	agent.WaitingFor = wait.ForHTTP("/").WithPort("8200/tcp").WithStartupTimeout(5 * time.Minute)
 	err = agent.Start()
 	require.NoError(t, err)
 
@@ -214,7 +223,7 @@ func cleanupFleet(t testing.TB, fleet *fleettest.Client) {
 		err = fleet.DeleteAgentPolicy(p.ID)
 		var fleetError *fleettest.Error
 		if errors.As(err, &fleetError) {
-			assert.Equal(t, http.StatusNotFound, fleetError.StatusCode)
+			assert.Equal(t, http.StatusNotFound, fleetError.StatusCode, fleetError.Message)
 		}
 	}
 }

systemtest/fleettest/client.go

@@ -70,13 +70,13 @@ func (c *Client) Agents() ([]Agent, error) {
 }
 
 // BulkUnenrollAgents bulk-unenrolls agents.
-func (c *Client) BulkUnenrollAgents(force bool, agentIDs ...string) error {
+func (c *Client) BulkUnenrollAgents(revoke bool, agentIDs ...string) error {
 	var body bytes.Buffer
 	type bulkUnenroll struct {
 		Agents []string `json:"agents"`
-		Force  bool     `json:"force"`
+		Revoke bool     `json:"revoke"`
 	}
-	if err := json.NewEncoder(&body).Encode(bulkUnenroll{agentIDs, force}); err != nil {
+	if err := json.NewEncoder(&body).Encode(bulkUnenroll{agentIDs, revoke}); err != nil {
 		return err
 	}
 	req := c.newFleetRequest("POST", "/agents/bulk_unenroll", &body)