Verify supported Elasticsearch distribution during license reconciliation

pkg/controller/elasticsearch/client/client.go

@@ -143,6 +143,14 @@ type APIError struct {
 	response *http.Response
 }
 
+func FakeAPIError(statusCode int) APIError {
+	return APIError{
+		response: &http.Response{
+			StatusCode: statusCode,
+		},
+	}
+}
+
 // Error() implements the error interface.
 func (e *APIError) Error() string {
 	defer e.response.Body.Close()
@@ -156,6 +164,11 @@ func (e *APIError) Error() string {
 	return fmt.Sprintf("%s: %s", e.response.Status, reason)
 }
 
+// IsUnauthorized checks whether the error was an HTTP 401 error.
+func IsUnauthorized(err error) bool {
+	return isHTTPError(err, http.StatusUnauthorized)
+}
+
 // IsForbidden checks whether the error was an HTTP 403 error.
 func IsForbidden(err error) bool {
 	return isHTTPError(err, http.StatusForbidden)
@@ -176,6 +189,15 @@ func IsConflict(err error) bool {
 	return isHTTPError(err, http.StatusConflict)
 }
 
+func Is4xx(err error) bool {
+	apiErr := new(APIError)
+	if errors.As(err, &apiErr) {
+		code := apiErr.response.StatusCode
+		return code >= 400 && code <= 499
+	}
+	return false
+}
+
 func isHTTPError(err error, statusCode int) bool {
 	apiErr := new(APIError)
 	if errors.As(err, &apiErr) {

pkg/controller/elasticsearch/driver/driver.go

@@ -8,8 +8,10 @@ import (
 	"context"
 	"crypto/x509"
 	"fmt"
+	"strings"
 	"time"
 
+	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/record"
 	controller "sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -195,15 +197,19 @@ func (d *defaultDriver) Reconcile(ctx context.Context) *reconciler.Results {
 	}
 
 	if esReachable {
-		// reconcile the license
-		if err := license.Reconcile(ctx, d.Client, d.ES, esClient); err != nil {
-			d.ReconcileState.AddEvent(
-				corev1.EventTypeWarning,
-				events.EventReasonUnexpected,
-				fmt.Sprintf("Could not update cluster license: %s", err.Error()),
-			)
-			log.Info("Could not update cluster license", "err", err, "namespace", d.ES.Namespace, "es_name", d.ES.Name)
-			// don't error out the entire reconciliation, move on with next steps and retry later
+		// reconcile the Elasticsearch license
+		supportedDistribution, err := license.Reconcile(ctx, d.Client, d.ES, esClient)
+		if err != nil && !supportedDistribution {
+			msg := "Unsupported Elasticsearch distribution"
+			d.ReconcileState.AddEvent(corev1.EventTypeWarning, events.EventReasonUnexpected, fmt.Sprintf("%s: %s", msg, err.Error()))
+			// unsupported distribution, let's update the phase to "invalid" and stop the reconciliation
+			d.ReconcileState.UpdateElasticsearchStatusPhase(esv1.ElasticsearchResourceInvalid)
+			return results.WithError(errors.Wrap(err, strings.ToLower(msg[0:1])+msg[1:]))
+		}
+		if err != nil {
+			msg := "Could not reconcile cluster license"
+			d.ReconcileState.AddEvent(corev1.EventTypeWarning, events.EventReasonUnexpected, fmt.Sprintf("%s: %s", msg, err.Error()))
+			log.Info(msg, "err", err, "namespace", d.ES.Namespace, "es_name", d.ES.Name)
 			results.WithResult(defaultRequeue)
 		}
 

pkg/controller/elasticsearch/license/apply.go

@@ -37,20 +37,15 @@ func applyLinkedLicense(
 	c k8s.Client,
 	esCluster types.NamespacedName,
 	updater esclient.LicenseClient,
+	currentLicense esclient.License,
 ) error {
-	// get the current license
-	current, err := updater.GetLicense(ctx)
-	if err != nil {
-		return fmt.Errorf("while getting current license level %w", err)
-	}
-
 	// get the expected license
 	// the underlying assumption here is that either a user or a
 	// license controller has created a cluster license in the
 	// namespace of this cluster following the cluster-license naming
 	// convention
 	var license corev1.Secret
-	err = c.Get(context.Background(),
+	err := c.Get(context.Background(),
 		types.NamespacedName{
 			Namespace: esCluster.Namespace,
 			Name:      esv1.LicenseSecretName(esCluster.Name),
@@ -63,10 +58,10 @@ func applyLinkedLicense(
 	if err != nil && apierrors.IsNotFound(err) {
 		// no license expected, let's look at the current cluster license
 		switch {
-		case isBasic(current):
+		case isBasic(currentLicense):
 			// nothing to do
 			return nil
-		case isTrial(current):
+		case isTrial(currentLicense):
 			// Elasticsearch reports a trial license, but there's no ECK enterprise trial requested.
 			// This can be the case if:
 			// - an ECK trial was started previously, then stopped (secret removed)
@@ -92,7 +87,7 @@ func applyLinkedLicense(
 	if err != nil {
 		return pkgerrors.Wrap(err, "no valid license found in license secret")
 	}
-	return updateLicense(ctx, esCluster, updater, current, desired)
+	return updateLicense(ctx, esCluster, updater, currentLicense, desired)
 }
 
 func startBasic(ctx context.Context, updater esclient.LicenseClient) error {

pkg/controller/elasticsearch/license/apply_test.go

@@ -222,6 +222,7 @@ func Test_applyLinkedLicense(t *testing.T) {
 				c,
 				clusterName,
 				&updater,
+				tt.currentLicense,
 			); (err != nil) != tt.wantErr {
 				t.Errorf("applyLinkedLicense() error = %v, wantErr %v", err, tt.wantErr)
 			}
@@ -232,6 +233,79 @@ func Test_applyLinkedLicense(t *testing.T) {
 	}
 }
 
+func Test_checkEsLicense(t *testing.T) {
+	tests := []struct {
+		name      string
+		wantErr   bool
+		supported bool
+		updater   esclient.LicenseClient
+	}{
+		{
+			name:      "happy path",
+			wantErr:   false,
+			supported: true,
+			updater: &fakeInvalidLicenseUpdater{
+				fakeLicenseUpdater:     &fakeLicenseUpdater{license: esclient.License{Type: string(esclient.ElasticsearchLicenseTypeBasic)}},
+				statusCodeOnGetLicense: 200,
+			},
+		},
+		{
+			name:      "error: 400 on get license, unsupported distribution",
+			wantErr:   true,
+			supported: false,
+			updater:   &fakeInvalidLicenseUpdater{statusCodeOnGetLicense: 400},
+		},
+		{
+			name:      "error: 401 on get license",
+			wantErr:   true,
+			supported: true,
+			updater:   &fakeInvalidLicenseUpdater{statusCodeOnGetLicense: 401},
+		},
+		{
+			name:      "error: 403 on get license",
+			wantErr:   true,
+			supported: true,
+			updater:   &fakeInvalidLicenseUpdater{statusCodeOnGetLicense: 403},
+		},
+		{
+			name:      "error: 404 on get license",
+			wantErr:   true,
+			supported: true,
+			updater:   &fakeInvalidLicenseUpdater{statusCodeOnGetLicense: 404},
+		},
+		{
+			name:      "error: 500 on get license",
+			wantErr:   true,
+			supported: true,
+			updater:   &fakeInvalidLicenseUpdater{statusCodeOnGetLicense: 500},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, supported, err := checkElasticsearchLicense(context.Background(), tt.updater)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("checkElasticsearchLicense() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if supported != tt.supported {
+				t.Errorf("checkElasticsearchLicense() supported = %v, supported %v", supported, tt.supported)
+			}
+		})
+	}
+}
+
+type fakeInvalidLicenseUpdater struct {
+	*fakeLicenseUpdater
+	statusCodeOnGetLicense int
+}
+
+func (f *fakeInvalidLicenseUpdater) GetLicense(ctx context.Context) (esclient.License, error) {
+	if f.statusCodeOnGetLicense == 200 {
+		return f.license, nil
+	}
+	apiErr := esclient.FakeAPIError(f.statusCodeOnGetLicense)
+	return esclient.License{}, &apiErr
+}
+
 type fakeLicenseUpdater struct {
 	license             esclient.License
 	startBasicCalled    bool

pkg/controller/elasticsearch/license/reconcile.go

@@ -10,6 +10,7 @@ import (
 	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
 	esclient "github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/client"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
+	"github.com/pkg/errors"
 )
 
 // Reconcile reconciles the current Elasticsearch license with the desired one.
@@ -18,7 +19,33 @@ func Reconcile(
 	c k8s.Client,
 	esCluster esv1.Elasticsearch,
 	clusterClient esclient.Client,
-) error {
+) (bool, error) {
+	currentLicense, supportedDistribution, err := checkElasticsearchLicense(ctx, clusterClient)
+	if err != nil {
+		return supportedDistribution, err
+	}
+
 	clusterName := k8s.ExtractNamespacedName(&esCluster)
-	return applyLinkedLicense(ctx, c, clusterName, clusterClient)
+	return true, applyLinkedLicense(ctx, c, clusterName, clusterClient, currentLicense)
+}
+
+// checkElasticsearchLicense checks that Elasticsearch is licensed, which ensures that the operator is communicating
+// with a supported Elasticsearch distribution
+func checkElasticsearchLicense(ctx context.Context, clusterClient esclient.LicenseClient) (esclient.License, bool, error) {
+	supportedDistribution := true
+	currentLicense, err := clusterClient.GetLicense(ctx)
+	if err != nil {
+		switch {
+		case esclient.IsUnauthorized(err):
+			err = errors.New("unauthorized access, unable to verify Elasticsearch license, check your security configuration")
+		case esclient.IsForbidden(err):
+			err = errors.New("forbidden access, unable to verify Elasticsearch license, check your security configuration")
+		case esclient.IsNotFound(err):
+			// 404 may happen if the master node is generating a new cluster state
+		case esclient.Is4xx(err):
+			supportedDistribution = false
+			err = errors.Wrap(err, "unable to verify Elasticsearch license")
+		}
+	}
+	return currentLicense, supportedDistribution, err
 }

pkg/controller/elasticsearch/reconcile/state.go

@@ -153,3 +153,7 @@ func (s *State) UpdateElasticsearchInvalid(err error) {
 	s.status.Phase = esv1.ElasticsearchResourceInvalid
 	s.AddEvent(corev1.EventTypeWarning, events.EventReasonValidation, err.Error())
 }
+
+func (s *State) UpdateElasticsearchStatusPhase(orchPhase esv1.ElasticsearchOrchestrationPhase) {
+	s.status.Phase = orchPhase
+}