functionaltests: Add support for interacting with v7 fleet (#16519)

* Add support for interacting with v7 fleet

* Update functionaltests/utils_test.go

Co-authored-by: Edoardo Tenani <[email protected]>

---------

Co-authored-by: Edoardo Tenani <[email protected]>

Author: ericywl

.github/workflows/ci.yml

@@ -57,7 +57,8 @@ jobs:
       - run: go test -v -race ./...
       - env:
           KIBANA_URL: "https://kibana.test"
-          KIBANA_API_KEY: "dummy"
+          KIBANA_USERNAME: "dummy"
+          KIBANA_PASSWORD: "dummy"
           EC_URL: "https://elastic-cloud.test"
           EC_API_KEY: "dummy"
           EC_REGION: "gcp-us-west2"

functionaltests/internal/gen/generator.go

@@ -135,7 +135,7 @@ func (g *Generator) RunBlockingWait(ctx context.Context, version ecclient.StackV
 	}
 
 	// With standalone, we don't have Fleet, so simply just wait for some arbitrary time.
-	time.Sleep(30 * time.Second)
+	time.Sleep(60 * time.Second)
 	return nil
 }
 
@@ -158,17 +158,14 @@ func flushAPMMetrics(ctx context.Context, kbc *kbclient.Client, version string)
 	// Sending an update with modifying the description is enough to trigger
 	// final aggregations in APM Server and flush of in-flight metrics.
 	policy.Description = fmt.Sprintf("Functional tests %s", version)
-	if err = kbc.UpdatePackagePolicyByID(ctx, policyID, kbclient.UpdatePackagePolicyRequest{
-		PackagePolicy: policy,
-		Force:         false,
-	}); err != nil {
+	if err = kbc.UpdatePackagePolicyByID(ctx, policyID, policy); err != nil {
 		return fmt.Errorf("cannot update elastic-cloud-apm package policy: %w", err)
 	}
 
 	// APM Server needs some time to flush all metrics, and we don't have any
 	// visibility on when this completes.
 	// NOTE: This value comes from empirical observations.
-	time.Sleep(20 * time.Second)
+	time.Sleep(40 * time.Second)
 	return nil
 }
 

functionaltests/internal/kbclient/client.go

@@ -30,17 +30,13 @@ import (
 	"time"
 )
 
-func New(kibanaURL, apiKey string, username, password string) (*Client, error) {
+func New(kibanaURL, username, password string) (*Client, error) {
 	if kibanaURL == "" {
 		return nil, fmt.Errorf("kbclient.New kibanaURL must not be empty")
 	}
-	if apiKey == "" {
-		return nil, fmt.Errorf("kbclient.New apiKey must not be empty")
-	}
 
 	return &Client{
 		url:                 kibanaURL,
-		apiKey:              apiKey,
 		superUsername:       username,
 		superPassword:       password,
 		SupportedAPIVersion: "2023-10-31",
@@ -52,9 +48,8 @@ type Client struct {
 	http.Client
 	// url is the Kibana URL where requests will be directed to.
 	url string
-	// apiKey should be an Elasticsearch API key with appropriate privileges.
-	apiKey string
 
+	// Fleet API access require superuser role before 8.1.0.
 	// superUsername should be an Elasticsearch superuser username.
 	superUsername string
 	// superPassword should be an Elasticsearch superuser password.
@@ -64,7 +59,7 @@ type Client struct {
 }
 
 // prepareRequest creates a http.Request with required headers for interacting with Kibana.
-func (c *Client) prepareRequest(method, path string, body any, super bool) (*http.Request, error) {
+func (c *Client) prepareRequest(method, path string, body any) (*http.Request, error) {
 	b, err := json.Marshal(body)
 	if err != nil {
 		return nil, fmt.Errorf("cannot marshal body: %w", err)
@@ -79,13 +74,10 @@ func (c *Client) prepareRequest(method, path string, body any, super bool) (*htt
 	req.Header.Add("kbn-xsrf", "true")
 	req.Header.Add("Content-Type", "application/json")
 	req.Header.Add("Elastic-Api-Version", c.SupportedAPIVersion)
-	if super {
-		userPass := fmt.Sprintf("%s:%s", c.superUsername, c.superPassword)
-		basicAuth := base64.StdEncoding.EncodeToString([]byte(userPass))
-		req.Header.Add("Authorization", fmt.Sprintf("Basic %s", basicAuth))
-	} else {
-		req.Header.Add("Authorization", fmt.Sprintf("ApiKey %s", c.apiKey))
-	}
+
+	userPass := fmt.Sprintf("%s:%s", c.superUsername, c.superPassword)
+	basicAuth := base64.StdEncoding.EncodeToString([]byte(userPass))
+	req.Header.Add("Authorization", fmt.Sprintf("Basic %s", basicAuth))
 
 	return req, nil
 }
@@ -98,11 +90,10 @@ func (c *Client) sendRequest(
 	method string,
 	path string,
 	body any,
-	super bool,
 	handleRespError func(statusCode int, body []byte) error,
 ) ([]byte, error) {
 	methodPath := fmt.Sprintf("%s %s", method, path)
-	req, err := c.prepareRequest(method, path, body, super)
+	req, err := c.prepareRequest(method, path, body)
 	if err != nil {
 		return nil, fmt.Errorf("cannot prepare request (%s): %w", methodPath, err)
 	}
@@ -144,18 +135,33 @@ func (e ElasticAgentPolicyNotFoundError) Error() string {
 }
 
 type PackagePolicy struct {
-	Name        string           `json:"name"`
-	Description string           `json:"description"`
-	Package     PackagePolicyPkg `json:"package"`
-	PolicyID    string           `json:"policy_id"`
-	PolicyIDs   []string         `json:"policy_ids"`
+	Name        string               `json:"name"`
+	Description string               `json:"description"`
+	Package     PackagePolicyPkg     `json:"package"`
+	Namespace   string               `json:"namespace"`
+	Inputs      []PackagePolicyInput `json:"inputs"`
+	OutputID    string               `json:"output_id"`
+	PolicyID    string               `json:"policy_id,omitempty"`
+	PolicyIDs   []string             `json:"policy_ids,omitempty"`
+	Enabled     bool                 `json:"enabled"`
 }
 
 type PackagePolicyPkg struct {
 	Name    string `json:"name"`
+	Title   string `json:"title"`
 	Version string `json:"version"`
 }
 
+type PackagePolicyInput struct {
+	Config         any            `json:"config"`
+	Enabled        bool           `json:"enabled"`
+	Streams        []any          `json:"streams"`
+	Type           string         `json:"type"`
+	Vars           map[string]any `json:"vars"`
+	KeepEnabled    bool           `json:"keep_enabled"`
+	PolicyTemplate string         `json:"policy_template"`
+}
+
 type getPackagePolicyResponse struct {
 	Item PackagePolicy `json:"item"`
 }
@@ -174,7 +180,7 @@ func (c *Client) GetPackagePolicyByID(ctx context.Context, policyID string) (Pac
 		return nil
 	}
 
-	b, err := c.sendRequest(ctx, http.MethodGet, path, nil, false, handleRespError)
+	b, err := c.sendRequest(ctx, http.MethodGet, path, nil, handleRespError)
 	if err != nil {
 		return PackagePolicy{}, err
 	}
@@ -187,19 +193,14 @@ func (c *Client) GetPackagePolicyByID(ctx context.Context, policyID string) (Pac
 	return policyResp.Item, nil
 }
 
-type UpdatePackagePolicyRequest struct {
-	PackagePolicy
-	Force bool `json:"force"`
-}
-
 // UpdatePackagePolicyByID performs a Package Policy update in Fleet through the Fleet Kibana APIs.
 // https://www.elastic.co/docs/api/doc/kibana/v8/operation/operation-update-package-policy
-func (c *Client) UpdatePackagePolicyByID(ctx context.Context, policyID string, request UpdatePackagePolicyRequest) error {
+func (c *Client) UpdatePackagePolicyByID(ctx context.Context, policyID string, policy PackagePolicy) error {
 	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
 	defer cancel()
 
 	path := fmt.Sprintf("/api/fleet/package_policies/%s", policyID)
-	_, err := c.sendRequest(ctx, http.MethodPut, path, request, false, nil)
+	_, err := c.sendRequest(ctx, http.MethodPut, path, policy, nil)
 	return err
 }
 
@@ -218,7 +219,7 @@ func (c *Client) EnableIntegrationsServer(ctx context.Context) error {
 	// This is an internal API that is not publicly documented.
 	// https://github.com/elastic/kibana/blob/12aa3fc/x-pack/solutions/observability/plugins/apm/server/routes/fleet/route.ts#L146
 	path := "/internal/apm/fleet/cloud_apm_package_policy"
-	b, err := c.sendRequest(ctx, http.MethodPost, path, nil, true, nil)
+	b, err := c.sendRequest(ctx, http.MethodPost, path, nil, nil)
 	if err != nil {
 		return err
 	}

functionaltests/internal/kbclient/client_test.go

@@ -110,8 +110,9 @@ func getHttpClient(t *testing.T) (*recorder.Recorder, *http.Client) {
 
 func TestClient_GetPackagePolicyByID(t *testing.T) {
 	kibanaURL := os.Getenv("KIBANA_URL")
-	apikey := os.Getenv("KIBANA_API_KEY")
-	kbc, err := kbclient.New(kibanaURL, apikey, "", "")
+	username := os.Getenv("KIBANA_USERNAME")
+	password := os.Getenv("KIBANA_PASSWORD")
+	kbc, err := kbclient.New(kibanaURL, username, password)
 	require.NoError(t, err)
 	_, httpc := getHttpClient(t)
 	kbc.Client = *httpc
@@ -123,25 +124,25 @@ func TestClient_GetPackagePolicyByID(t *testing.T) {
 
 func TestClient_UpdatePackagePolicyByID(t *testing.T) {
 	kibanaURL := os.Getenv("KIBANA_URL")
-	apiKey := os.Getenv("KIBANA_API_KEY")
-	kbc, err := kbclient.New(kibanaURL, apiKey, "", "")
+	username := os.Getenv("KIBANA_USERNAME")
+	password := os.Getenv("KIBANA_PASSWORD")
+	kbc, err := kbclient.New(kibanaURL, username, password)
 	require.NoError(t, err)
 	_, httpc := getHttpClient(t)
 	kbc.Client = *httpc
 
 	ctx := context.Background()
 	policyID := "elastic-cloud-apm"
-	err = kbc.UpdatePackagePolicyByID(ctx, policyID, kbclient.UpdatePackagePolicyRequest{
-		PackagePolicy: kbclient.PackagePolicy{
+	err = kbc.UpdatePackagePolicyByID(ctx, policyID,
+		kbclient.PackagePolicy{
 			Name:        "Elastic APM",
 			Description: "Hello World",
 			Package: kbclient.PackagePolicyPkg{
 				Name:    "apm",
 				Version: "9.1.0-SNAPSHOT",
 			},
 		},
-		Force: false,
-	})
+	)
 	require.NoError(t, err)
 
 	policy, err := kbc.GetPackagePolicyByID(ctx, policyID)

functionaltests/single_upgrade_test.go

@@ -80,7 +80,7 @@ func (tt singleUpgradeTestCase) Run(t *testing.T) {
 	t.Logf("time elapsed: %s", time.Since(start))
 
 	esc := createESClient(t, deployInfo)
-	kbc := createKibanaClient(t, ctx, esc, deployInfo)
+	kbc := createKibanaClient(t, deployInfo)
 	g := createAPMGenerator(t, ctx, esc, kbc, deployInfo)
 
 	atStartCount := getDocCountPerDS(t, ctx, esc)

functionaltests/steps_test.go

@@ -129,7 +129,7 @@ func (c createStep) Step(t *testing.T, ctx context.Context, e *testStepEnv, _ te
 	integrations := c.APMDeploymentMode.enableIntegrations()
 	deployInfo := createCluster(t, ctx, e.tf, *target, c.DeployVersion, integrations)
 	e.esc = createESClient(t, deployInfo)
-	e.kbc = createKibanaClient(t, ctx, e.esc, deployInfo)
+	e.kbc = createKibanaClient(t, deployInfo)
 	e.gen = createAPMGenerator(t, ctx, e.esc, e.kbc, deployInfo)
 	// Update the environment version to the new one.
 	e.version = c.DeployVersion

functionaltests/utils_test.go

@@ -169,13 +169,10 @@ func createESClient(t *testing.T, deployInfo deploymentInfo) *esclient.Client {
 }
 
 // createKibanaClient instantiate an HTTP API client with dedicated methods to query the Kibana API.
-// This function will also create an Elasticsearch API key with full permissions to be used by the HTTP client.
-func createKibanaClient(t *testing.T, ctx context.Context, esc *esclient.Client, deployInfo deploymentInfo) *kbclient.Client {
+func createKibanaClient(t *testing.T, deployInfo deploymentInfo) *kbclient.Client {
 	t.Helper()
 	t.Log("create kibana client")
-	apiKey, err := esc.CreateAPIKey(ctx, "kbclient", -1, map[string]types.RoleDescriptor{})
-	require.NoError(t, err)
-	kbc, err := kbclient.New(deployInfo.KibanaURL, apiKey, deployInfo.Username, deployInfo.Password)
+	kbc, err := kbclient.New(deployInfo.KibanaURL, deployInfo.Username, deployInfo.Password)
 	require.NoError(t, err)
 	return kbc
 }