Send fleet-server elasticsearch config under new bootstrap attribute (#4643)

Alter the fleet-server bootstrap component modifier to insert all (elasticsearch)
output configuration options specified by enrollment args under a new elasticsearch.boostrap key
instead of overwriting any existing keys. This will allow elastic-agent to send the list of hosts
(and other config options) retrieved from a policy to fleet server as well as the config needed
to form the initial connection to elasticsearch used to collect policy information.
Fleet-server has been altered to use the bootstrap config that is passed if the policy attributes
are unspecified or fail.

Author: michel-laterman

changelog/fragments/1714505172-Send-fleet-server-component-output-under-bootstrap-key.yaml

@@ -0,0 +1,36 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: enhancement
+
+# Change summary; a 80ish characters long description of the change.
+summary: Fleet Server component now uses policy output configuration to communicate with Elasticsearch
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+description: |
+  Alter how elatic-agent passes the fleet-server output component so that the policy's output is used.
+  In cases where fleet-server encounters an error when trying to use the policy's output it will use
+  the configuration specified during enrollment as a fallback. In cases where it uses the fallback
+  the policy's output is periodically retested and used if it's successful.
+
+# Affected component; a word indicating the component this changeset affects.
+component:
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: https://github.com/elastic/elastic-agent/pull/4643
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: https://github.com/elastic/elastic-agent/issue/2784

docs/fleet-server-bootstrap.asciidoc

@@ -88,3 +88,61 @@ its API key to use for communication. The new `fleet.yml` still includes the `fl
 but this time the `fleet.server.bootstrap: false` is set.
 . `enroll` command then either restarts the running Elatic Agent daemon if one was running
 from Step 2, or it stops the spawned `run` subprocess and returns.
+
+=== Elasticsearch output
+
+The options passed that are used to specify fleet-server initially connects to elasticsearch are:
+
+- `--fleet-server-es`
+- `--fleet-server-es-ca`
+- `--fleet-server-es-ca-trusted-fingerprint`
+- `--fleet-server-es-insecure`
+- `--fleet-server-es-cert`
+- `--fleet-server-es-cert-key`
+- `--fleet-server-es-service-token`
+- `--fleet-server-es-service-token-path`
+- `--proxy-url`
+- `--proxy-disabled`
+- `--proxy-header`
+
+These options are always passed under a `bootstrap` attribute in the output when elastic-agent is passing config to fleet-server.
+When the fleet-server recieves an output block, it will inject any keys that are missing from the top level output but are specified in the `bootstrap` block
+After injecting the keys from bootstrap, fleet-server will test connecting the Elasticsearch with the output.
+If the test fails, the values under the `bootstrap` attribute are used as the output and fleet-server will periodically retest the output in case the error was caused by a temporary network issue.
+Note that if `--fleet-server-es-insecure` is specified, and the output in the policy contains one or more CA, or a CA fingerprint, the `--fleet-server-es-insecure` flag is ignored.
+
+An example of this sequence is sequence is:
+
+1) elastic-agent starts fleet-server and sends an output block that looks similar to:
+```yaml
+output:
+  bootstrap:
+    service_token: VALUE
+    hosts: ["HOST"]
+```
+
+2) fleet-server injects attributes into the top level from bootstrap if they are missing, resulting in
+```yaml
+output:
+  service_token: VALUE
+  hosts: ["HOST"]
+```
+
+3) fleet-server connects to Elasticsearch with the output block
+4) elastic-agent enrolls and recieves its policy
+5) elastic-agent sends configuration generated from the policy to fleet-server, this may result in the output as follows:
+```yaml
+output:
+  hosts: ["HOST", "HOST2"]
+  bootstrap:
+    service_token: VALUE
+    hosts: ["HOST"]
+```
+
+6) fleet-server will inject missing values resulting in:
+```yaml
+output:
+  service_token: VALUE
+  hosts: ["HOST", "HOST2"]
+```
+ 7) fleet-server tests and uses the resulting output block.

internal/pkg/agent/application/fleet_server_bootstrap.go

@@ -65,10 +65,13 @@ func FleetServerComponentModifier(serverCfg *configuration.FleetServerConfig) co
 				} else {
 					for j, unit := range comp.Units {
 						if unit.Type == client.UnitTypeOutput && unit.Config.Type == elasticsearch {
-							unitCfgMap, err := toMapStr(unit.Config.Source.AsMap(), &serverCfg.Output.Elasticsearch)
+							unitCfgMap, err := toMapStr(unit.Config.Source.AsMap())
 							if err != nil {
 								return nil, err
 							}
+							if err := addBootstrapCfg(unitCfgMap, &serverCfg.Output.Elasticsearch); err != nil {
+								return nil, err
+							}
 							fixOutputMap(unitCfgMap)
 							unitCfg, err := component.ExpectedConfig(unitCfgMap)
 							if err != nil {
@@ -100,6 +103,19 @@ func FleetServerComponentModifier(serverCfg *configuration.FleetServerConfig) co
 	}
 }
 
+// addBootrapCfg will transform the passed configuration.Elasticsearch to a map and add it to dst under the bootstrap key.
+func addBootstrapCfg(dst map[string]interface{}, es *configuration.Elasticsearch) error {
+	if es == nil {
+		return fmt.Errorf("fleet-server bootstrap output config is undefined")
+	}
+	mp, err := toMapStr(es)
+	if err != nil {
+		return err
+	}
+	dst["bootstrap"] = mp
+	return nil
+}
+
 // InjectFleetConfigComponentModifier The modifier that injects the fleet configuration for the components
 // that need to be able to connect to fleet server.
 func InjectFleetConfigComponentModifier(fleetCfg *configuration.FleetAgentConfig, agentInfo info.Agent) coordinator.ComponentsModifier {

internal/pkg/agent/application/fleet_server_bootstrap_test.go

@@ -74,6 +74,87 @@ func TestFleetServerComponentModifier_NoServerConfig(t *testing.T) {
 	}
 }
 
+func TestFleetServerComponentModifier(t *testing.T) {
+	tests := []struct {
+		name   string
+		source map[string]interface{}
+		expect map[string]interface{}
+	}{{
+		name:   "empty output component",
+		source: map[string]interface{}{},
+		expect: map[string]interface{}{
+			"bootstrap": map[string]interface{}{
+				"protocol":      "https",
+				"hosts":         []interface{}{"elasticsearch:9200"},
+				"service_token": "example-token",
+			},
+		},
+	}, {
+		name: "output component provided",
+		source: map[string]interface{}{
+			"protocol": "http",
+			"hosts":    []interface{}{"elasticsearch:9200", "host:9200"},
+		},
+		expect: map[string]interface{}{
+			"protocol": "http",
+			"hosts":    []interface{}{"elasticsearch:9200", "host:9200"},
+			"bootstrap": map[string]interface{}{
+				"protocol":      "https",
+				"hosts":         []interface{}{"elasticsearch:9200"},
+				"service_token": "example-token",
+			},
+		},
+	}}
+	cfg := &configuration.FleetServerConfig{
+		Output: configuration.FleetServerOutputConfig{
+			Elasticsearch: configuration.Elasticsearch{
+				Protocol:     "https",
+				Hosts:        []string{"elasticsearch:9200"},
+				ServiceToken: "example-token",
+			},
+		},
+	}
+	modifier := FleetServerComponentModifier(cfg)
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			src, err := structpb.NewStruct(tc.source)
+			require.NoError(t, err)
+			comps, err := modifier([]component.Component{{
+				InputSpec: &component.InputRuntimeSpec{
+					InputType: "fleet-server",
+				},
+				Units: []component.Unit{{
+					Type: client.UnitTypeOutput,
+					Config: &proto.UnitExpectedConfig{
+						Type:   "elasticsearch",
+						Source: src,
+					},
+				}},
+			}}, nil)
+			require.NoError(t, err)
+
+			require.Len(t, comps, 1)
+			require.Len(t, comps[0].Units, 1)
+			res := comps[0].Units[0].Config.Source.AsMap()
+			for k, v := range tc.expect {
+				val, ok := res[k]
+				require.Truef(t, ok, "expected %q to be in output unit config", k)
+				if mp, ok := v.(map[string]interface{}); ok {
+					rMap, ok := val.(map[string]interface{})
+					require.Truef(t, ok, "expected %q to be map[string]interface{} was %T", k, val)
+					for kk, vv := range mp {
+						assert.Contains(t, rMap, kk)
+						assert.Equal(t, rMap[kk], vv)
+					}
+				} else {
+					assert.Equal(t, v, val)
+				}
+			}
+		})
+	}
+}
+
 func TestInjectFleetConfigComponentModifier(t *testing.T) {
 	fleetConfig := &configuration.FleetAgentConfig{
 		Enabled: true,

internal/pkg/agent/configuration/fleet_server.go

@@ -32,7 +32,14 @@ type FleetServerOutputConfig struct {
 	Elasticsearch Elasticsearch `config:"elasticsearch" yaml:"elasticsearch"`
 }
 
-// Elasticsearch is the configuration for elasticsearch.
+// Elasticsearch is the configuration for fleet-server's connection to elasticsearch.
+// Note that these keys may be injected into policy output by fleet-server.
+// The following TLS options may be set in bootstrap:
+// - VerificationMode
+// - CAs
+// - CATrustedFingerprint
+// - CertificateConfig.Certificate AND CertificateConfig.Key
+// If an attribute is added to this struct, or another TLS attribute is passed  ensure that it is handled as part of the bootstrap config handler in fleet-server/internal/pkg/server/agent.go
 type Elasticsearch struct {
 	Protocol         string            `config:"protocol" yaml:"protocol"`
 	Hosts            []string          `config:"hosts" yaml:"hosts"`

pkg/testing/fixture_install.go

@@ -56,6 +56,30 @@ func (e EnrollOpts) toCmdArgs() []string {
 	return args
 }
 
+type FleetBootstrapOpts struct {
+	ESHost       string // --fleet-server-es
+	ServiceToken string // --fleet-server-service-token
+	Policy       string // --fleet-server-policy
+	Port         int    // --fleet-server-port
+}
+
+func (f FleetBootstrapOpts) toCmdArgs() []string {
+	var args []string
+	if f.ESHost != "" {
+		args = append(args, "--fleet-server-es", f.ESHost)
+	}
+	if f.ServiceToken != "" {
+		args = append(args, "--fleet-server-service-token", f.ServiceToken)
+	}
+	if f.Policy != "" {
+		args = append(args, "--fleet-server-policy", f.Policy)
+	}
+	if f.Port > 0 {
+		args = append(args, "--fleet-server-port", fmt.Sprintf("%d", f.Port))
+	}
+	return args
+}
+
 // InstallOpts specifies the options for the install command
 type InstallOpts struct {
 	BasePath       string // --base-path
@@ -68,6 +92,7 @@ type InstallOpts struct {
 	Privileged bool // inverse of --unprivileged (as false is the default)
 
 	EnrollOpts
+	FleetBootstrapOpts
 }
 
 func (i InstallOpts) toCmdArgs(operatingSystem string) ([]string, error) {
@@ -95,6 +120,7 @@ func (i InstallOpts) toCmdArgs(operatingSystem string) ([]string, error) {
 	}
 
 	args = append(args, i.EnrollOpts.toCmdArgs()...)
+	args = append(args, i.FleetBootstrapOpts.toCmdArgs()...)
 
 	return args, nil
 }

pkg/testing/tools/estools/elasticsearch.go

@@ -13,6 +13,8 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/google/uuid"
+
 	"github.com/elastic/elastic-agent-libs/mapstr"
 	"github.com/elastic/elastic-transport-go/v8/elastictransport"
 	"github.com/elastic/go-elasticsearch/v8/esapi"
@@ -201,6 +203,35 @@ func CreateAPIKey(ctx context.Context, client elastictransport.Interface, req AP
 	return parsed, nil
 }
 
+func CreateServiceToken(ctx context.Context, client elastictransport.Interface, service string) (string, error) {
+	req := esapi.SecurityCreateServiceTokenRequest{
+		Namespace: "elastic",
+		Service:   service,
+		Name:      uuid.New().String(), // FIXME(michel-laterman): We need to specify a random name until an upstream issue is fixed: https://github.com/elastic/go-elasticsearch/issues/861
+	}
+	resp, err := req.Do(ctx, client)
+	if err != nil {
+		return "", fmt.Errorf("error creating service token: %w", err)
+	}
+	defer resp.Body.Close()
+	resultBuf, err := handleResponseRaw(resp)
+	if err != nil {
+		return "", fmt.Errorf("error handling HTTP response: %w", err)
+	}
+
+	var parsed struct {
+		Token struct {
+			Value string `json:"value"`
+		} `json:"token"`
+	}
+	err = json.Unmarshal(resultBuf, &parsed)
+	if err != nil {
+		return "", fmt.Errorf("error unmarshaling json response: %w", err)
+	}
+	return parsed.Token.Value, nil
+
+}
+
 // FindMatchingLogLines returns any logs with message fields that match the given line
 func FindMatchingLogLines(ctx context.Context, client elastictransport.Interface, namespace, line string) (Documents, error) {
 	return FindMatchingLogLinesWithContext(ctx, client, namespace, line)

testing/integration/fleet-server.json

@@ -0,0 +1,38 @@
+{
+    "id": "3434b864-d135-4d03-a944-29ee7ad61ddd",
+    "version": "WzMwNywxXQ==",
+    "name": "fleet_server-1",
+    "namespace": "",
+    "description": "",
+    "package": {
+      "name": "fleet_server",
+      "title": "Fleet Server",
+      "version": "1.5.0"
+    },
+    "enabled": true,
+    "inputs": [
+      {
+        "type": "fleet-server",
+        "policy_template": "fleet_server",
+        "enabled": true,
+        "streams": [],
+        "vars": {
+          "max_agents": {
+            "type": "integer"
+          },
+          "max_connections": {
+            "type": "integer"
+          },
+          "custom": {
+            "value": "",
+            "type": "yaml"
+          }
+        }
+      }
+    ],
+    "revision": 1,
+    "created_at": "2024-05-22T16:13:09.177Z",
+    "created_by": "system",
+    "updated_at": "2024-05-22T16:13:09.177Z",
+    "updated_by": "system"
+}