Add local K8S remote cluster support (#2543)

* Add local K8S remote cluster support

cmd/manager/main.go

@@ -46,6 +46,7 @@ import (
 	kbassn "github.com/elastic/cloud-on-k8s/pkg/controller/kibanaassociation"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/license"
 	licensetrial "github.com/elastic/cloud-on-k8s/pkg/controller/license/trial"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/remoteca"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/webhook"
 	"github.com/elastic/cloud-on-k8s/pkg/dev"
 	"github.com/elastic/cloud-on-k8s/pkg/dev/portforward"
@@ -336,6 +337,10 @@ func execute() {
 		log.Error(err, "unable to create controller", "controller", "KibanaAssociation")
 		os.Exit(1)
 	}
+	if err = remoteca.Add(mgr, accessReviewer, params); err != nil {
+		log.Error(err, "unable to create controller", "controller", "RemoteClusterCertificateAuthorites")
+		os.Exit(1)
+	}
 
 	if err = license.Add(mgr, params); err != nil {
 		log.Error(err, "unable to create controller", "controller", "License")

config/crds/all-crds.yaml

@@ -1043,6 +1043,37 @@ spec:
                       type: object
                   type: object
               type: object
+            remoteClusters:
+              description: RemoteClusters enables you to establish uni-directional
+                connections to a remote Elasticsearch cluster.
+              items:
+                description: RemoteCluster declares a remote Elasticsearch cluster
+                  connection.
+                properties:
+                  elasticsearchRef:
+                    description: ElasticsearchRef is a reference to an Elasticsearch
+                      cluster running within the same k8s cluster.
+                    properties:
+                      name:
+                        description: Name of the Kubernetes object.
+                        type: string
+                      namespace:
+                        description: Namespace of the Kubernetes object. If empty,
+                          defaults to the current namespace.
+                        type: string
+                    required:
+                    - name
+                    type: object
+                  name:
+                    description: Name is the name of the remote cluster as it is set
+                      in the Elasticsearch settings. The name is expected to be unique
+                      for each remote clusters.
+                    minLength: 1
+                    type: string
+                required:
+                - name
+                type: object
+              type: array
             secureSettings:
               description: 'SecureSettings is a list of references to Kubernetes secrets
                 containing sensitive configuration options for Elasticsearch. See:

config/crds/bases/elasticsearch.k8s.elastic.co_elasticsearches.yaml

@@ -6859,6 +6859,37 @@ spec:
                         type: object
                     type: object
                 type: object
+              remoteClusters:
+                description: RemoteClusters enables you to establish uni-directional
+                  connections to a remote Elasticsearch cluster.
+                items:
+                  description: RemoteCluster declares a remote Elasticsearch cluster
+                    connection.
+                  properties:
+                    elasticsearchRef:
+                      description: ElasticsearchRef is a reference to an Elasticsearch
+                        cluster running within the same k8s cluster.
+                      properties:
+                        name:
+                          description: Name of the Kubernetes object.
+                          type: string
+                        namespace:
+                          description: Namespace of the Kubernetes object. If empty,
+                            defaults to the current namespace.
+                          type: string
+                      required:
+                      - name
+                      type: object
+                    name:
+                      description: Name is the name of the remote cluster as it is
+                        set in the Elasticsearch settings. The name is expected to
+                        be unique for each remote clusters.
+                      minLength: 1
+                      type: string
+                  required:
+                  - name
+                  type: object
+                type: array
               secureSettings:
                 description: 'SecureSettings is a list of references to Kubernetes
                   secrets containing sensitive configuration options for Elasticsearch.

docs/reference/api-docs.asciidoc

@@ -197,6 +197,7 @@ ObjectSelector defines a reference to a Kubernetes object.
 ****
 - xref:{anchor_prefix}-github-com-elastic-cloud-on-k8s-pkg-apis-apm-v1-apmserverspec[$$ApmServerSpec$$]
 - xref:{anchor_prefix}-github-com-elastic-cloud-on-k8s-pkg-apis-kibana-v1-kibanaspec[$$KibanaSpec$$]
+- xref:{anchor_prefix}-github-com-elastic-cloud-on-k8s-pkg-apis-elasticsearch-v1-remotecluster[$$RemoteCluster$$]
 ****
 
 [cols="25a,75a", options="header"]
@@ -616,6 +617,7 @@ ElasticsearchSpec holds the specification of an Elasticsearch cluster.
 | *`podDisruptionBudget`* __xref:{anchor_prefix}-github-com-elastic-cloud-on-k8s-pkg-apis-common-v1-poddisruptionbudgettemplate[$$PodDisruptionBudgetTemplate$$]__ | PodDisruptionBudget provides access to the default pod disruption budget for the Elasticsearch cluster. The default budget selects all cluster pods and sets `maxUnavailable` to 1. To disable, set `PodDisruptionBudget` to the empty value (`{}` in YAML).
 | *`secureSettings`* __xref:{anchor_prefix}-github-com-elastic-cloud-on-k8s-pkg-apis-common-v1-secretsource[$$SecretSource$$]__ | SecureSettings is a list of references to Kubernetes secrets containing sensitive configuration options for Elasticsearch. See: https://www.elastic.co/guide/en/cloud-on-k8s/current/k8s-es-secure-settings.html
 | *`serviceAccountName`* __string__ | ServiceAccountName is used to check access from the current resource to a resource (eg. a remote Elasticsearch cluster) in a different namespace. Can only be used if ECK is enforcing RBAC on references.
+| *`remoteClusters`* __xref:{anchor_prefix}-github-com-elastic-cloud-on-k8s-pkg-apis-elasticsearch-v1-remotecluster[$$RemoteCluster$$] array__ | RemoteClusters enables you to establish uni-directional connections to a remote Elasticsearch cluster.
 |===
 
 
@@ -642,6 +644,24 @@ ElasticsearchSpec holds the specification of an Elasticsearch cluster.
 |===
 
 
+[id="{anchor_prefix}-github-com-elastic-cloud-on-k8s-pkg-apis-elasticsearch-v1-remotecluster"]
+=== RemoteCluster 
+
+
+
+.Appears In:
+****
+- xref:{anchor_prefix}-github-com-elastic-cloud-on-k8s-pkg-apis-elasticsearch-v1-elasticsearchspec[$$ElasticsearchSpec$$]
+****
+
+[cols="25a,75a", options="header"]
+|===
+| Field | Description
+| *`name`* __string__ | Name is the name of the remote cluster as it is set in the Elasticsearch settings. The name is expected to be unique for each remote clusters.
+| *`elasticsearchRef`* __xref:{anchor_prefix}-github-com-elastic-cloud-on-k8s-pkg-apis-common-v1-objectselector[$$ObjectSelector$$]__ | ElasticsearchRef is a reference to an Elasticsearch cluster running within the same k8s cluster.
+|===
+
+
 [id="{anchor_prefix}-github-com-elastic-cloud-on-k8s-pkg-apis-elasticsearch-v1-updatestrategy"]
 === UpdateStrategy 
 

pkg/apis/common/v1/common.go

@@ -33,18 +33,29 @@ type ObjectSelector struct {
 	Namespace string `json:"namespace,omitempty"`
 }
 
+// WithDefaultNamespace adds a default namespace to a given ObjectSelector if none is set.
+func (o ObjectSelector) WithDefaultNamespace(defaultNamespace string) ObjectSelector {
+	if len(o.Namespace) > 0 {
+		return o
+	}
+	return ObjectSelector{
+		Namespace: defaultNamespace,
+		Name:      o.Name,
+	}
+}
+
 // NamespacedName is a convenience method to turn an ObjectSelector into a NamespacedName.
-func (s ObjectSelector) NamespacedName() types.NamespacedName {
+func (o ObjectSelector) NamespacedName() types.NamespacedName {
 	return types.NamespacedName{
-		Name:      s.Name,
-		Namespace: s.Namespace,
+		Name:      o.Name,
+		Namespace: o.Namespace,
 	}
 }
 
 // IsDefined checks if the object selector is not nil and has a name.
 // Namespace is not mandatory as it may be inherited by the parent object.
-func (s *ObjectSelector) IsDefined() bool {
-	return s != nil && s.Name != ""
+func (o *ObjectSelector) IsDefined() bool {
+	return o != nil && o.Name != ""
 }
 
 // HTTPConfig holds the HTTP layer configuration for resources.

pkg/apis/elasticsearch/v1/elasticsearch_types.go

@@ -5,6 +5,7 @@
 package v1
 
 import (
+	"github.com/elastic/cloud-on-k8s/pkg/controller/common/hash"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
@@ -50,6 +51,29 @@ type ElasticsearchSpec struct {
 	// Can only be used if ECK is enforcing RBAC on references.
 	// +optional
 	ServiceAccountName string `json:"serviceAccountName,omitempty"`
+
+	// RemoteClusters enables you to establish uni-directional connections to a remote Elasticsearch cluster.
+	// +optional
+	RemoteClusters []RemoteCluster `json:"remoteClusters,omitempty"`
+}
+
+// RemoteCluster declares a remote Elasticsearch cluster connection.
+type RemoteCluster struct {
+	// Name is the name of the remote cluster as it is set in the Elasticsearch settings.
+	// The name is expected to be unique for each remote clusters.
+	// +kubebuilder:validation:Required
+	// +kubebuilder:validation:MinLength=1
+	Name string `json:"name"`
+
+	// ElasticsearchRef is a reference to an Elasticsearch cluster running within the same k8s cluster.
+	ElasticsearchRef commonv1.ObjectSelector `json:"elasticsearchRef,omitempty"`
+
+	// TODO: Allow the user to specify some options (transport.compress, transport.ping_schedule)
+
+}
+
+func (r RemoteCluster) ConfigHash() string {
+	return hash.HashObject(r)
 }
 
 // NodeCount returns the total number of nodes of the Elasticsearch cluster

pkg/apis/elasticsearch/v1/name.go

@@ -18,6 +18,7 @@ const (
 	configSecretSuffix                = "config"
 	secureSettingsSecretSuffix        = "secure-settings"
 	httpServiceSuffix                 = "http"
+	transportServiceSuffix            = "transport"
 	elasticUserSecretSuffix           = "elastic-user"
 	xpackFileRealmSecretSuffix        = "xpack-file-realm"
 	internalUsersSecretSuffix         = "internal-users"
@@ -27,6 +28,9 @@ const (
 	scriptsConfigMapSuffix            = "scripts"
 	transportCertificatesSecretSuffix = "transport-certificates"
 
+	// remoteCaNameSuffix is a suffix for the secret that contains the concatenation of all the remote CAs
+	remoteCaNameSuffix = "remote-ca"
+
 	controllerRevisionHashLen = 10
 )
 
@@ -46,6 +50,7 @@ var (
 		defaultPodDisruptionBudget,
 		scriptsConfigMapSuffix,
 		transportCertificatesSecretSuffix,
+		remoteCaNameSuffix,
 	}
 )
 
@@ -108,6 +113,10 @@ func TransportCertificatesSecret(esName string) string {
 	return ESNamer.Suffix(esName, transportCertificatesSecretSuffix)
 }
 
+func TransportService(esName string) string {
+	return ESNamer.Suffix(esName, transportServiceSuffix)
+}
+
 func HTTPService(esName string) string {
 	return ESNamer.Suffix(esName, httpServiceSuffix)
 }
@@ -140,3 +149,7 @@ func LicenseSecretName(esName string) string {
 func DefaultPodDisruptionBudget(esName string) string {
 	return ESNamer.Suffix(esName, defaultPodDisruptionBudget)
 }
+
+func RemoteCaSecretName(esName string) string {
+	return ESNamer.Suffix(esName, remoteCaNameSuffix)
+}

pkg/controller/elasticsearch/certificates/ca_reconcile.go

@@ -15,6 +15,7 @@ import (
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/driver"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/reconciler"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/tracing"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/certificates/remoteca"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/certificates/transport"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/label"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
@@ -48,6 +49,11 @@ func Reconcile(
 
 	results := &reconciler.Results{}
 
+	// reconcile remote clusters certificate authorities
+	if err := remoteca.Reconcile(driver.K8sClient(), es); err != nil {
+		results.WithError(err)
+	}
+
 	labels := label.NewLabels(k8s.ExtractNamespacedName(&es))
 
 	httpCA, err := certificates.ReconcileCAForOwner(

pkg/controller/elasticsearch/certificates/remoteca/reconcile.go

@@ -0,0 +1,78 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package remoteca
+
+import (
+	"bytes"
+	"reflect"
+	"sort"
+
+	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/common/certificates"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/common/reconciler"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/label"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/remotecluster/remoteca"
+	"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
+	"github.com/elastic/cloud-on-k8s/pkg/utils/maps"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// Reconcile fetches the list of remote certificate authorities and concatenates them into a single Secret
+func Reconcile(
+	c k8s.Client,
+	es esv1.Elasticsearch,
+) error {
+	// Get all the remote certificate authorities
+	var remoteCAList v1.SecretList
+	if err := c.List(
+		&remoteCAList,
+		client.InNamespace(es.Namespace),
+		remoteca.LabelSelector(es.Name),
+	); err != nil {
+		return err
+	}
+	// We sort the remote certificate authorities to have a stable comparison with the reconciled data
+	sort.SliceStable(remoteCAList.Items, func(i, j int) bool {
+		// We don't need to compare the namespace because they are all in the same one
+		return remoteCAList.Items[i].Name < remoteCAList.Items[j].Name
+	})
+
+	remoteCertificateAuthorities := make([][]byte, len(remoteCAList.Items))
+	for i, remoteCA := range remoteCAList.Items {
+		remoteCertificateAuthorities[i] = remoteCA.Data[certificates.CAFileName]
+	}
+
+	expected := v1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      esv1.RemoteCaSecretName(es.Name),
+			Namespace: es.Namespace,
+			Labels: map[string]string{
+				label.ClusterNameLabelName: es.Name,
+			},
+		},
+		Data: map[string][]byte{
+			certificates.CAFileName: bytes.Join(remoteCertificateAuthorities, nil),
+		},
+	}
+
+	var reconciled v1.Secret
+	return reconciler.ReconcileResource(reconciler.Params{
+		Client:     c,
+		Scheme:     scheme.Scheme,
+		Owner:      &es,
+		Expected:   &expected,
+		Reconciled: &reconciled,
+		NeedsUpdate: func() bool {
+			return !maps.IsSubset(expected.Labels, reconciled.Labels) || !reflect.DeepEqual(expected.Data, reconciled.Data)
+		},
+		UpdateReconciled: func() {
+			reconciled.Labels = maps.Merge(reconciled.Labels, expected.Labels)
+			reconciled.Data = expected.Data
+		},
+	})
+}