Allow Elastic Agent in different namespace than Elastic Stack.

.buildkite/pipeline.yml

@@ -11,7 +11,7 @@ steps:
         agents:
           image: docker.elastic.co/ci-agent-images/cloud-k8s-operator/buildkite-agent:bfddf2b3
           cpu: "6"
-          memory: "6G"
+          memory: "7G"
 
       - label: ":go: generate"
         command: "make generate check-local-changes"

docs/orchestrating-elastic-stack-applications/agent-fleet.asciidoc

@@ -702,12 +702,15 @@ Deploys single instance Elastic Agent Deployment in Fleet mode with APM integrat
 [id="{p}-elastic-agent-fleet-known-limitations"]
 == Known limitations
 
-=== Running as root and within a single namespace (ECK < 2.10.0 and Agent < 7.14.0)
-Until version 7.14.0 and ECK version 2.10.0, Elastic Agent in Fleet mode has to run as root and in the same namespace as the Elasticsearch cluster it connects to. 
+=== Running as root (ECK < 2.10.0 and Agent < 7.14.0)
+Until version 7.14.0 and ECK version 2.10.0, Elastic Agent and Fleet Server were required to run as root.
 
-This was due to configuration limitations in Fleet/Elastic Agent. ECK needed to establish trust between Elastic Agents and Elasticsearch. ECK was only able to fetch the required Elasticsearch CA correctly if both resources are in the same namespace.
 As of Elastic Stack version 7.14.0 and ECK version 2.10.0 it is also possible to run Elastic Agent and Fleet as a non-root user. See <<{p}_storing_local_state_in_host_path_volume>> for instructions.
-To establish trust, the Pod needs to update the CA store through a call to `update-ca-trust` before Elastic Agent runs. To call it successfully, the Pod needs to run with elevated privileges.
+
+=== Elastic Agent running in the same namespace as Elastic Fleet Server
+Until ECK version 2.11.0, Elastic Agent and Fleet Server were required to run within the same Namespace.
+
+As of ECK version 2.11.0, Elastic Agent and Fleet Server can be deployed in different Namespaces.
 
 === Running Endpoint Security integration
 Running Endpoint Security link:https://www.elastic.co/guide/en/security/current/install-endpoint.html[integration] is not yet supported in containerized environments, like Kubernetes. This is not an ECK limitation, but the limitation of the integration itself. Note that you can use ECK to deploy Elasticsearch, Kibana and Fleet Server, and add Endpoint Security integration to your policies if Elastic Agents running those policies are deployed in non-containerized environments.

pkg/controller/agent/driver.go

@@ -108,7 +108,13 @@ func internalReconcile(params Params) (*reconciler.Results, agentv1alpha1.AgentS
 
 	configHash := fnv.New32a()
 	var fleetCerts *certificates.CertificatesSecret
-	if params.Agent.Spec.FleetServerEnabled && params.Agent.Spec.HTTP.TLS.Enabled() {
+	if params.Agent.Spec.HTTP.TLS.Enabled() && params.Agent.Spec.FleetModeEnabled() {
+		// Only Fleet Server has a Service associated with it,
+		// Fleet-enabled Agents do not have a Service.
+		var services []corev1.Service
+		if svc != nil {
+			services = append(services, *svc)
+		}
 		var caResults *reconciler.Results
 		fleetCerts, caResults = certificates.Reconciler{
 			K8sClient:                   params.Client,
@@ -117,7 +123,7 @@ func internalReconcile(params Params) (*reconciler.Results, agentv1alpha1.AgentS
 			TLSOptions:                  params.Agent.Spec.HTTP.TLS,
 			Namer:                       Namer,
 			Labels:                      params.Agent.GetIdentityLabels(),
-			Services:                    []corev1.Service{*svc},
+			Services:                    services,
 			GlobalCA:                    params.OperatorParams.GlobalCA,
 			CACertRotation:              params.OperatorParams.CACertRotation,
 			CertRotation:                params.OperatorParams.CertRotation,

pkg/controller/agent/pod.go

@@ -302,6 +302,7 @@ func getRelatedEsAssoc(params Params) (commonv1.Association, error) {
 			return nil, err
 		}
 	} else if params.Agent.Spec.FleetServerRef.IsDefined() {
+		agent := params.Agent
 		// As the reference chain is: Elastic Agent ---> Fleet Server ---> Elasticsearch,
 		// we need first to identify the Fleet Server and then identify its reference to Elasticsearch.
 		fsAssociation, err := association.SingleAssociationOfType(params.Agent.GetAssociations(), commonv1.FleetServerAssociationType)
@@ -320,7 +321,11 @@ func getRelatedEsAssoc(params Params) (commonv1.Association, error) {
 			return nil, pkgerrors.Wrap(err, "while fetching associated fleet server")
 		}
 
-		esAssociation, err = association.SingleAssociationOfType(fs.GetAssociations(), commonv1.ElasticsearchAssociationType)
+		// We copy the Fleet Server Refs to the Agent so that the association appears to come from
+		// the Elastic Agent, not the Fleet Server and is named appropriately.
+		agent.Spec.ElasticsearchRefs = fs.Spec.ElasticsearchRefs
+
+		esAssociation, err = association.SingleAssociationOfType(agent.GetAssociations(), commonv1.ElasticsearchAssociationType)
 		if err != nil {
 			return nil, err
 		}
@@ -333,26 +338,30 @@ func applyRelatedEsAssoc(agent agentv1alpha1.Agent, esAssociation commonv1.Assoc
 		return builder, nil
 	}
 
-	esRef := esAssociation.AssociationRef()
-	if !esRef.IsExternal() && !agent.Spec.FleetServerEnabled && agent.Namespace != esRef.Namespace {
-		// check agent and ES share the same namespace
-		return nil, fmt.Errorf(
-			"agent namespace %s is different than referenced Elasticsearch namespace %s, this is not supported yet",
-			agent.Namespace,
-			esAssociation.AssociationRef().Namespace,
-		)
-	}
-
 	// no ES CA to configure, skip
 	assocConf, err := esAssociation.AssociationConf()
 	if err != nil {
 		return nil, err
 	}
-	if !assocConf.CAIsConfigured() {
+
+	// A transitive association is an association that is not directly configured by the user but is created
+	// by associating a Fleet-enabled Agent with a Fleet Server. The transitive association in that case
+	// will be Elastic Agent => Fleet-Server => Elasticsearch.
+	transitiveAssociation := isTransitiveAssociation(agent, esAssociation)
+	if !assocConf.CAIsConfigured() && !transitiveAssociation {
 		return builder, nil
 	}
+	// If the association configuration has the CA configuration directly in the annotation
+	// then we can simply use the secret specified in the annotation.
+	caSecretName := assocConf.GetCASecretName()
+	if transitiveAssociation {
+		// In the case of a transitive association, no CA is configured in the annotation, so we need to
+		// use the method used within the Association controller to generate the expected secret name.
+		caSecretName = association.CACertSecretName(esAssociation, "agent-fleetserver")
+	}
+
 	builder = builder.WithVolumeLikes(volume.NewSecretVolumeWithMountPath(
-		assocConf.GetCASecretName(),
+		caSecretName,
 		fmt.Sprintf("%s-certs", esAssociation.AssociationType()),
 		certificatesDir(esAssociation),
 	))
@@ -371,6 +380,15 @@ func applyRelatedEsAssoc(agent agentv1alpha1.Agent, esAssociation commonv1.Assoc
 	return builder, nil
 }
 
+// isTransitiveAssociation returns true if the given association is a transitive association, which is defined
+// as an association that is not directly configured by the user but is created by associating a Fleet-enabled
+// Agent with a Fleet Server which indirectly associates Elastic Agent with Elasticsearch.
+func isTransitiveAssociation(agent agentv1alpha1.Agent, association commonv1.Association) bool {
+	return association.AssociationType() == commonv1.ElasticsearchAssociationType &&
+		agent.Spec.FleetModeEnabled() &&
+		agent.Spec.FleetServerRef.IsDefined()
+}
+
 func runningAsRoot(agent agentv1alpha1.Agent) bool {
 	if agent.Spec.DaemonSet != nil {
 		return runningContainerAsRoot(agent.Spec.DaemonSet.PodTemplate)

pkg/controller/agent/pod_test.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 
 	"github.com/go-test/deep"
+	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -834,6 +835,10 @@ func Test_applyRelatedEsAssoc(t *testing.T) {
 	})
 
 	assocToOtherNs := (&agentv1alpha1.Agent{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "es-agent",
+			Namespace: agentNs,
+		},
 		Spec: agentv1alpha1.AgentSpec{
 			ElasticsearchRefs: []agentv1alpha1.Output{
 				{
@@ -857,13 +862,31 @@ func Test_applyRelatedEsAssoc(t *testing.T) {
 			},
 		},
 	}
+	expectedFleetCAVolume := []corev1.Volume{
+		{
+			Name: "elasticsearch-certs",
+			VolumeSource: corev1.VolumeSource{
+				Secret: &corev1.SecretVolumeSource{
+					SecretName: "es-agent-agent-fleetserver-elasticsearch-ns-elasticsearch-ca",
+					Optional:   &optional,
+				},
+			},
+		},
+	}
 	expectedCAVolumeMount := []corev1.VolumeMount{
 		{
 			Name:      "elasticsearch-certs",
 			ReadOnly:  true,
 			MountPath: "/mnt/elastic-internal/elasticsearch-association/agent-ns/elasticsearch/certs",
 		},
 	}
+	expectedFleetCAVolumeMount := []corev1.VolumeMount{
+		{
+			Name:      "elasticsearch-certs",
+			ReadOnly:  true,
+			MountPath: "/mnt/elastic-internal/elasticsearch-association/elasticsearch-ns/elasticsearch/certs",
+		},
+	}
 	expectedCmd := []string{"/usr/bin/env", "bash", "-c", `#!/usr/bin/env bash
 set -e
 if [[ -f /mnt/elastic-internal/elasticsearch-association/agent-ns/elasticsearch/certs/ca.crt ]]; then
@@ -958,7 +981,59 @@ fi
 				},
 			},
 			assoc:   assocToOtherNs,
-			wantErr: true,
+			wantErr: false,
+			wantPodSpec: generatePodSpec(func(ps corev1.PodSpec) corev1.PodSpec {
+				ps.Volumes = nil
+				ps.Containers[0].VolumeMounts = nil
+				ps.Containers[0].Command = nil
+				return ps
+			}),
+		},
+		{
+			name: "fleet server disabled, fleet reference in place in same namespace, esref in different namespace",
+			agent: agentv1alpha1.Agent{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "agent",
+					Namespace: agentNs,
+				},
+				Spec: agentv1alpha1.AgentSpec{
+					Mode:               agentv1alpha1.AgentFleetMode,
+					FleetServerEnabled: false,
+					FleetServerRef:     commonv1.ObjectSelector{Name: "fs", Namespace: agentNs},
+					Version:            "7.16.2",
+				},
+			},
+			assoc:   assocToOtherNs,
+			wantErr: false,
+			wantPodSpec: generatePodSpec(func(ps corev1.PodSpec) corev1.PodSpec {
+				ps.Volumes = expectedFleetCAVolume
+				ps.Containers[0].VolumeMounts = expectedFleetCAVolumeMount
+				ps.Containers[0].Command = nil
+				return ps
+			}),
+		},
+		{
+			name: "fleet server disabled, fleet reference in place in different namespace, esref in different namespace",
+			agent: agentv1alpha1.Agent{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "agent",
+					Namespace: agentNs,
+				},
+				Spec: agentv1alpha1.AgentSpec{
+					Mode:               agentv1alpha1.AgentFleetMode,
+					FleetServerEnabled: false,
+					FleetServerRef:     commonv1.ObjectSelector{Name: "fs", Namespace: "other-ns"},
+					Version:            "7.16.2",
+				},
+			},
+			assoc:   assocToOtherNs,
+			wantErr: false,
+			wantPodSpec: generatePodSpec(func(ps corev1.PodSpec) corev1.PodSpec {
+				ps.Volumes = expectedFleetCAVolume
+				ps.Containers[0].VolumeMounts = expectedFleetCAVolumeMount
+				ps.Containers[0].Command = nil
+				return ps
+			}),
 		},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
@@ -967,7 +1042,7 @@ fi
 			require.Equal(t, tt.wantErr, gotErr != nil)
 			if !tt.wantErr {
 				require.Nil(t, gotErr)
-				require.Nil(t, deep.Equal(tt.wantPodSpec, gotBuilder.PodTemplate.Spec))
+				require.Nil(t, deep.Equal(tt.wantPodSpec, gotBuilder.PodTemplate.Spec), "wantPodSpec != got, diff: %s", cmp.Diff(tt.wantPodSpec, gotBuilder.PodTemplate.Spec))
 			}
 		})
 	}

pkg/controller/association/ca.go

@@ -16,7 +16,6 @@ import (
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/certificates"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/name"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/reconciler"
-	"github.com/elastic/cloud-on-k8s/v2/pkg/utils/k8s"
 )
 
 // CASecret is a container to hold information about the Elasticsearch CA secret.
@@ -46,7 +45,7 @@ func (r *Reconciler) ReconcileCASecret(ctx context.Context, association commonv1
 		return CASecret{}, err
 	}
 
-	labels := r.AssociationResourceLabels(k8s.ExtractNamespacedName(association), association.AssociationRef().NamespacedName())
+	labels := r.AssociationResourceLabels(association)
 
 	// Certificate data should be copied over a secret in the association namespace
 	expectedSecret := corev1.Secret{

pkg/controller/association/controller/agent_fleetserver.go

@@ -42,9 +42,35 @@ func AddAgentFleetServer(mgr manager.Manager, accessReviewer rbac.AccessReviewer
 		AssociationResourceNamespaceLabelName: agent.NamespaceLabelName,
 
 		ElasticsearchUserCreation: nil,
+
+		TransitivelyAssociated: getFleetAssociatedResources,
 	})
 }
 
+func getFleetAssociatedResources(c k8s.Client, assoc commonv1.Association) ([]commonv1.Associated, error) {
+	associated := assoc.Associated()
+	agent := agentv1alpha1.Agent{}
+	nsn := types.NamespacedName{Namespace: associated.GetNamespace(), Name: associated.GetName()}
+	if err := c.Get(context.Background(), nsn, &agent); err != nil {
+		return nil, err
+	}
+	fleetServerRef := assoc.AssociationRef()
+	if !fleetServerRef.IsDefined() {
+		return nil, nil
+	}
+	fleetServer := agentv1alpha1.Agent{}
+	if err := c.Get(context.Background(), fleetServerRef.NamespacedName(), &fleetServer); err != nil {
+		return nil, err
+	}
+	// If the Fleet Server Agent is not associated with an Elasticsearch cluster
+	// (potentially because of a manual setup) we should do nothing.
+	if len(fleetServer.Spec.ElasticsearchRefs) == 0 {
+		return []commonv1.Associated{}, nil
+	}
+	agent.Spec.ElasticsearchRefs = fleetServer.Spec.ElasticsearchRefs
+	return []commonv1.Associated{&agent}, nil
+}
+
 func getFleetServerExternalURL(c k8s.Client, assoc commonv1.Association) (string, error) {
 	fleetServerRef := assoc.AssociationRef()
 	if !fleetServerRef.IsDefined() {

pkg/controller/association/dynamic_watches.go

@@ -9,6 +9,7 @@ import (
 
 	"k8s.io/apimachinery/pkg/types"
 
+	agentv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/agent/v1alpha1"
 	commonv1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/common/v1"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/certificates"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/watches"
@@ -62,7 +63,12 @@ func (r *Reconciler) reconcileWatches(associated types.NamespacedName, associati
 	}
 
 	// watch the CA secret of the referenced resource in the referenced resource namespace
-	if err := ReconcileWatch(associated, managedElasticRef, r.watches.Secrets, referencedResourceCASecretWatchName(associated), func(association commonv1.Association) types.NamespacedName {
+	// and potentially the CA secret of the transitively associated resources.
+	managedElasticRefWithTransitives, err := r.maybeAddTransitiveAssociations(managedElasticRef, associations)
+	if err != nil {
+		return err
+	}
+	if err := ReconcileWatch(associated, managedElasticRefWithTransitives, r.watches.Secrets, referencedResourceCASecretWatchName(associated), func(association commonv1.Association) types.NamespacedName {
 		ref := association.AssociationRef()
 		return types.NamespacedName{
 			Name:      certificates.PublicCertsSecretName(r.AssociationInfo.ReferencedResourceNamer, ref.NameOrSecretName()),
@@ -96,6 +102,35 @@ func (r *Reconciler) reconcileWatches(associated types.NamespacedName, associati
 	return nil
 }
 
+// maybeAddTransitiveAssociations potentially adds the Elasticsearch instance's association associated with the Fleet Server
+// so we can watch the CA secret of the Elasticsearch instance and reconcile if it changes.
+func (r *Reconciler) maybeAddTransitiveAssociations(managedElasticRef, associations []commonv1.Association) ([]commonv1.Association, error) {
+	managedElasticRefWithTransitives := make([]commonv1.Association, len(managedElasticRef))
+	copy(managedElasticRefWithTransitives, managedElasticRef)
+	if r.AssociationInfo.AssociationType == commonv1.FleetServerAssociationType {
+		// add the ES instance's CAs associated with the fleet server
+		for _, assoc := range associations {
+			associatedFleetResources, err := r.AssociationInfo.TransitivelyAssociated(r.Client, assoc)
+			if err != nil {
+				return nil, err
+			}
+			if len(associatedFleetResources) == 0 {
+				continue
+			}
+			agent, ok := associatedFleetResources[0].(*agentv1alpha1.Agent)
+			if !ok {
+				continue
+			}
+			esAssociation, err := SingleAssociationOfType(agent.GetAssociations(), commonv1.ElasticsearchAssociationType)
+			if err != nil {
+				return nil, err
+			}
+			managedElasticRefWithTransitives = append(managedElasticRefWithTransitives, esAssociation) //nolint:makezero
+		}
+	}
+	return managedElasticRefWithTransitives, nil
+}
+
 // ReconcileWatch sets or removes `watchName` watch in `dynamicRequest` based on `associated` and `associations` and
 // `watchedFunc`. No watch is added if watchedFunc(association) refers to an empty namespaced name.
 func ReconcileWatch(