Do not use service accounts until Elasticsearch nodes have been upgraded (#5830) (#5831)

* Detect if all nodes are running with service accounts

* Do not use service accounts until Elasticsearch supports them

Author: barkbay

pkg/apis/elasticsearch/v1/elasticsearch_types.go

@@ -7,6 +7,7 @@ package v1
 import (
 	"strings"
 
+	"github.com/blang/semver/v4"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
@@ -41,6 +42,19 @@ const (
 	Kind = "Elasticsearch"
 )
 
+// ServiceAccountMinVersion is the first version of Elasticsearch for which ECK supports service accounts.
+// It is however up to each association controller to ensure that a specific service account is available
+// in the current Elasticsearch version.
+var ServiceAccountMinVersion = semver.MustParse("7.17.0")
+
+func AreServiceAccountsSupported(version string) (bool, error) {
+	esVersion, err := semver.Parse(version)
+	if err != nil {
+		return false, err
+	}
+	return esVersion.GTE(ServiceAccountMinVersion), nil
+}
+
 // +kubebuilder:object:root=true
 
 // ElasticsearchList contains a list of Elasticsearch clusters

pkg/controller/association/reconciler.go

@@ -32,6 +32,7 @@ import (
 	"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/common/watches"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/hints"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/user"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/maps"
@@ -330,10 +331,23 @@ func (r *Reconciler) reconcileAssociation(ctx context.Context, association commo
 	if err != nil {
 		return commonv1.AssociationPending, err
 	}
-	// Detect if we should use a service account. If it is the case create the related Secrets and update the association
-	// configuration on the associated resource.
-	assocLabels := r.AssociationResourceLabels(k8s.ExtractNamespacedName(association.Associated()), assocRef.NamespacedName())
+	// Detect if we should use a service account.
+	var esHints hints.OrchestrationsHints
 	if len(serviceAccount) > 0 {
+		// We must first ensure that the relevant orchestration hint is set on the Elasticsearch cluster.
+		esHints, err = hints.NewFrom(es)
+		if err != nil {
+			return commonv1.AssociationPending, err
+		}
+		if !esHints.ServiceAccounts.IsSet() {
+			r.log(k8s.ExtractNamespacedName(association)).Info("Waiting for Elasticsearch to report if service accounts are fully rolled out")
+			return commonv1.AssociationPending, nil
+		}
+	}
+
+	// If it is the case create the related Secrets and update the association configuration on the associated resource.
+	assocLabels := r.AssociationResourceLabels(k8s.ExtractNamespacedName(association.Associated()), assocRef.NamespacedName())
+	if len(serviceAccount) > 0 && esHints.ServiceAccounts.IsTrue() {
 		applicationSecretName := secretKey(association, r.ElasticsearchUserCreation.UserSecretSuffix)
 		r.log(k8s.ExtractNamespacedName(association)).V(1).Info("Ensure service account exists", "sa", serviceAccount)
 		err := ReconcileServiceAccounts(

pkg/controller/elasticsearch/client/client.go

@@ -60,6 +60,7 @@ type Client interface {
 	DesiredNodesClient
 	ShardLister
 	LicenseClient
+	SecurityClient
 	// Close idle connections in the underlying http client.
 	Close()
 	// Equal returns true if other can be considered as the same client.

pkg/controller/elasticsearch/client/security.go

@@ -0,0 +1,54 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+package client
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/elastic/cloud-on-k8s/pkg/utils/set"
+)
+
+type ServiceAccountCredential struct {
+	NodesCredentials NodesCredentials `json:"nodes_credentials"`
+}
+
+type NodesCredentials struct {
+	FileTokens map[string]FileToken `json:"file_tokens"`
+}
+
+type FileToken struct {
+	Nodes []string `json:"nodes"`
+}
+
+// Nodes returns the list of nodes which are referenced in the API response.
+func (s *ServiceAccountCredential) Nodes() set.StringSet {
+	result := set.Make()
+	for _, fileToken := range s.NodesCredentials.FileTokens {
+		for _, nodeName := range fileToken.Nodes {
+			result.Add(nodeName)
+		}
+	}
+	return result
+}
+
+type SecurityClient interface {
+
+	// GetServiceAccountCredentials returns the service account credentials from the /_security/service API
+	GetServiceAccountCredentials(ctx context.Context, namespacedService string) (ServiceAccountCredential, error)
+}
+
+func (c *clientV6) GetServiceAccountCredentials(_ context.Context, _ string) (ServiceAccountCredential, error) {
+	return ServiceAccountCredential{}, errNotSupportedInEs6x
+}
+
+func (c *clientV7) GetServiceAccountCredentials(ctx context.Context, namespacedService string) (ServiceAccountCredential, error) {
+	var serviceAccountCredential ServiceAccountCredential
+	path := fmt.Sprintf("/_security/service/%s/credential", namespacedService)
+	if err := c.get(ctx, path, &serviceAccountCredential); err != nil {
+		return serviceAccountCredential, err
+	}
+	return serviceAccountCredential, nil
+}

pkg/controller/elasticsearch/client/security_test.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 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+package client
+
+import (
+	"context"
+	"io/ioutil"
+	"net/http"
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/cloud-on-k8s/pkg/controller/common/version"
+)
+
+func Test_GetServiceAccountCredentials(t *testing.T) {
+	type args struct {
+		namespacedService string
+	}
+	tests := []struct {
+		name    string
+		client  Client
+		args    args
+		want    ServiceAccountCredential
+		wantErr bool
+	}{
+		{
+			args: args{namespacedService: "elastic/kibana"},
+			want: ServiceAccountCredential{NodesCredentials: NodesCredentials{
+				FileTokens: map[string]FileToken{
+					"default_kibana-sample_50d2f2d2-d989-4ab7-a3d4-c9e31e5651ca": {Nodes: []string{"elasticsearch-sample-es-default-0"}},
+				}},
+			},
+			client: NewMockClient(version.MustParse("7.17.0"), func(req *http.Request) *http.Response {
+				require.Equal(t, "/_security/service/elastic/kibana/credential", req.URL.Path)
+				return &http.Response{
+					StatusCode: 200,
+					Body: ioutil.NopCloser(strings.NewReader(
+						`{
+	"service_account": "elastic/kibana",
+	"count": 1,
+	"tokens": {},
+	"nodes_credentials": {
+		"_nodes": {
+			"total": 1,
+			"successful": 1,
+			"failed": 0
+		},
+		"file_tokens": {
+			"default_kibana-sample_50d2f2d2-d989-4ab7-a3d4-c9e31e5651ca": {
+				"nodes": ["elasticsearch-sample-es-default-0"]
+			}
+		}
+	}
+}`)),
+					Header:  make(http.Header),
+					Request: req,
+				}
+			}),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := tt.client.GetServiceAccountCredentials(context.TODO(), tt.args.namespacedService)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("clientV7.GetServiceAccountCredentials() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("client.GetServiceAccountCredentials() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/controller/elasticsearch/driver/driver.go

@@ -14,6 +14,7 @@ import (
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/utils/pointer"
 	controller "sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
@@ -33,6 +34,7 @@ import (
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/cleanup"
 	esclient "github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/client"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/configmap"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/hints"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/initcontainer"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/label"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/license"
@@ -44,6 +46,8 @@ import (
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/stackmon"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/user"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
+	"github.com/elastic/cloud-on-k8s/pkg/utils/optional"
+	"github.com/elastic/cloud-on-k8s/pkg/utils/set"
 )
 
 var (
@@ -257,6 +261,10 @@ func (d *defaultDriver) Reconcile(ctx context.Context) *reconciler.Results {
 		}
 	}
 
+	// Update the service account orchestration hint. This is done early in the reconciliation loop to unblock association
+	// controllers that may be waiting for the orchestration hint.
+	results.WithError(d.maybeSetServiceAccountsOrchestrationHint(ctx, esReachable, esClient, resourcesState))
+
 	// reconcile the Elasticsearch license
 	if esReachable {
 		err = license.Reconcile(ctx, d.Client, d.ES, esClient, currentLicense)
@@ -354,6 +362,112 @@ func (d *defaultDriver) newElasticsearchClient(
 	)
 }
 
+// maybeSetServiceAccountsOrchestrationHint attempts to update an orchestration hint to let the association controllers
+// know whether all the nodes in the cluster are ready to authenticate service accounts.
+func (d *defaultDriver) maybeSetServiceAccountsOrchestrationHint(
+	ctx context.Context,
+	esReachable bool,
+	securityClient esclient.SecurityClient,
+	resourcesState *reconcile.ResourcesState,
+) error {
+	if d.ReconcileState.OrchestrationHints().ServiceAccounts.IsTrue() {
+		// Orchestration hint is already set to true, there is no point going back to false.
+		return nil
+	}
+
+	// Case 1: New cluster, we can immediately set the orchestration hint.
+	if !bootstrap.AnnotatedForBootstrap(d.ES) {
+		allNodesRunningServiceAccounts, err := esv1.AreServiceAccountsSupported(d.ES.Spec.Version)
+		if err != nil {
+			return err
+		}
+		d.ReconcileState.UpdateOrchestrationHints(
+			d.ReconcileState.OrchestrationHints().Merge(hints.OrchestrationsHints{ServiceAccounts: optional.NewBool(allNodesRunningServiceAccounts)}),
+		)
+		return nil
+	}
+
+	// Case 2: This is an existing cluster, but actual cluster version does not support service accounts.
+	if d.ES.Status.Version == "" {
+		return nil
+	}
+	supportServiceAccounts, err := esv1.AreServiceAccountsSupported(d.ES.Status.Version)
+	if err != nil {
+		return err
+	}
+	if !supportServiceAccounts {
+		d.ReconcileState.UpdateOrchestrationHints(
+			d.ReconcileState.OrchestrationHints().Merge(hints.OrchestrationsHints{ServiceAccounts: optional.NewBool(false)}),
+		)
+		return nil
+	}
+
+	// Case 3: cluster is already running with a version that does support service account and tokens have already been created.
+	// We don't however know if all nodes have been migrated and are running with the service_tokens file mounted from the configuration Secret.
+	// Let's try to detect that situation by comparing the existing nodes and the ones returned by the /_security/service API.
+	// Note that starting with release 2.3 the association controller does not create the service account token until Elasticsearch is annotated
+	// as compatible with service accounts. This is mostly to unblock situation described in https://github.com/elastic/cloud-on-k8s/issues/5684
+	if !esReachable {
+		// This requires the Elasticsearch API to be available
+		return nil
+	}
+	allPods := names(resourcesState.AllPods)
+	// Detect if some service tokens are expected
+	saTokens, err := user.GetServiceAccountTokens(d.Client, d.ES)
+	if err != nil {
+		log.Info("Could not detect if service accounts are expected", "err", err, "namespace", d.ES.Namespace, "es_name", d.ES.Name)
+		return err
+	}
+
+	allNodesRunningServiceAccounts, err := allNodesRunningServiceAccounts(ctx, saTokens, set.Make(allPods...), securityClient)
+	if err != nil {
+		log.Info("Could not detect if all nodes are ready for using service accounts", "err", err, "namespace", d.ES.Namespace, "es_name", d.ES.Name)
+		return err
+	}
+	if allNodesRunningServiceAccounts != nil {
+		d.ReconcileState.UpdateOrchestrationHints(
+			d.ReconcileState.OrchestrationHints().Merge(hints.OrchestrationsHints{ServiceAccounts: optional.NewBool(*allNodesRunningServiceAccounts)}),
+		)
+	}
+
+	return nil
+}
+
+// allNodesRunningServiceAccounts attempts to detect if all the nodes in the clusters have loaded the service_tokens file.
+// It returns nil if no decision can be made, for example when there is no tokens are expected to be found.
+func allNodesRunningServiceAccounts(
+	ctx context.Context,
+	saTokens user.ServiceAccountTokens,
+	allPods set.StringSet,
+	securityClient esclient.SecurityClient,
+) (*bool, error) {
+	if len(allPods) == 0 {
+		return nil, nil
+	}
+	if len(saTokens) == 0 {
+		// No tokens are expected: we cannot call the Elasticsearch API to detect which nodes are
+		// running with the conf/service_tokens file.
+		return nil, nil
+	}
+
+	// Get the namespaced service name to call the /_security/service/<namespace>/<service>/credential API
+	namespacedServices := saTokens.NamespacedServices()
+
+	// Get the nodes which have loaded tokens from the conf/service_tokens file.
+	for namespacedService := range namespacedServices {
+		credentials, err := securityClient.GetServiceAccountCredentials(ctx, namespacedService)
+		if err != nil {
+			return nil, err
+		}
+		diff := allPods.Diff(credentials.Nodes())
+		if len(diff) == 0 {
+			return pointer.Bool(true), nil
+		}
+	}
+	// Some nodes are running but did not show up in the security API.
+	return pointer.Bool(false), nil
+}
+
 // warnUnsupportedDistro sends an event of type warning if the Elasticsearch Docker image is not a supported
 // distribution by looking at if the prepare fs init container terminated with the UnsupportedDistro exit code.
 func warnUnsupportedDistro(pods []corev1.Pod, recorder *events.Recorder) {