Avoid reporting outdated ES health on reconciliation error that prevents getting the real one

pkg/controller/common/certificates/reconcile.go

@@ -8,7 +8,6 @@ import (
 	"context"
 	"time"
 
-	"go.elastic.co/apm"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
@@ -18,7 +17,6 @@ import (
 	commonv1 "github.com/elastic/cloud-on-k8s/pkg/apis/common/v1"
 	commonname "github.com/elastic/cloud-on-k8s/pkg/controller/common/name"
 	"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/utils/k8s"
 	ulog "github.com/elastic/cloud-on-k8s/pkg/utils/log"
@@ -53,9 +51,6 @@ type Reconciler struct {
 // - a Secret containing the public-facing HTTP certificates (same as the internal one, but without the key)
 // If TLS is disabled, self-signed certificates are still reconciled, for simplicity/consistency, but not used.
 func (r Reconciler) ReconcileCAAndHTTPCerts(ctx context.Context) (*CertificatesSecret, *reconciler.Results) {
-	span, _ := apm.StartSpan(ctx, "reconcile_certs", tracing.SpanTypeApp)
-	defer span.End()
-
 	results := reconciler.NewResult(ctx)
 
 	if !r.TLSOptions.Enabled() && r.GarbageCollectSecrets {

pkg/controller/elasticsearch/certificates/reconcile.go

@@ -35,16 +35,16 @@ type CertificateResources struct {
 	TransportCA *certificates.CA
 }
 
-// Reconcile reconciles the certificates of a cluster.
-func Reconcile(
+// Reconcile reconciles the HTTP layer certificates of a cluster.
+func ReconcileHTTP(
 	ctx context.Context,
 	driver driver.Interface,
 	es esv1.Elasticsearch,
 	services []corev1.Service,
 	caRotation certificates.RotationParams,
 	certRotation certificates.RotationParams,
-) (*CertificateResources, *reconciler.Results) {
-	span, _ := apm.StartSpan(ctx, "reconcile_certs", tracing.SpanTypeApp)
+) ([]*x509.Certificate, *reconciler.Results) {
+	span, _ := apm.StartSpan(ctx, "reconcile_http_certs", tracing.SpanTypeApp)
 	defer span.End()
 
 	var results *reconciler.Results
@@ -82,6 +82,30 @@ func Reconcile(
 		return nil, results
 	}
 
+	trustedHTTPCertificates, err := certificates.ParsePEMCerts(httpCerts.CertPem())
+	if err != nil {
+		return nil, results.WithError(err)
+	}
+
+	return trustedHTTPCertificates, nil
+}
+
+// Reconcile reconciles the transport layer certificates of a cluster.
+func ReconcileTransport(
+	ctx context.Context,
+	driver driver.Interface,
+	es esv1.Elasticsearch,
+	caRotation certificates.RotationParams,
+	certRotation certificates.RotationParams,
+) (*certificates.CA, *reconciler.Results) {
+	span, _ := apm.StartSpan(ctx, "reconcile_transport_certs", tracing.SpanTypeApp)
+	defer span.End()
+
+	results := reconciler.NewResult(ctx)
+
+	// label certificates secrets with the cluster name
+	certsLabels := label.NewLabels(k8s.ExtractNamespacedName(&es))
+
 	// reconcile transport CA and certs
 	transportCA, err := transport.ReconcileOrRetrieveCA(
 		driver,
@@ -119,13 +143,5 @@ func Reconcile(
 		return nil, results
 	}
 
-	trustedHTTPCertificates, err := certificates.ParsePEMCerts(httpCerts.CertPem())
-	if err != nil {
-		return nil, results.WithError(err)
-	}
-
-	return &CertificateResources{
-		TrustedHTTPCertificates: trustedHTTPCertificates,
-		TransportCA:             transportCA,
-	}, results
+	return transportCA, results
 }

pkg/controller/elasticsearch/driver/driver.go

@@ -115,6 +115,10 @@ var _ commondriver.Interface = &defaultDriver{}
 func (d *defaultDriver) Reconcile(ctx context.Context) *reconciler.Results {
 	results := reconciler.NewResult(ctx)
 
+	// reset the Elasticsearch health to 'unknown' so that if reconciliation fails before the observer has had a chance to get it,
+	// we stop reporting a state that may be out of date
+	d.ReconcileState.UpdateElasticsearchPendingUnknownHealth()
+
 	// garbage collect secrets attached to this cluster that we don't need anymore
 	if err := cleanup.DeleteOrphanedSecrets(ctx, d.Client, d.ES); err != nil {
 		return results.WithError(err)
@@ -140,54 +144,68 @@ func (d *defaultDriver) Reconcile(ctx context.Context) *reconciler.Results {
 		return results.WithError(err)
 	}
 
-	certificateResources, res := certificates.Reconcile(
+	resourcesState, err := reconcile.NewResourcesStateFromAPI(d.Client, d.ES)
+	if err != nil {
+		return results.WithError(err)
+	}
+
+	warnUnsupportedDistro(resourcesState.AllPods, d.ReconcileState.Recorder)
+
+	controllerUser, err := user.ReconcileUsersAndRoles(ctx, d.Client, d.ES, d.DynamicWatches(), d.Recorder())
+	if err != nil {
+		return results.WithError(err)
+	}
+
+	trustedHTTPCertificates, res := certificates.ReconcileHTTP(
 		ctx,
 		d,
 		d.ES,
 		[]corev1.Service{*externalService, *internalService},
 		d.OperatorParameters.CACertRotation,
 		d.OperatorParameters.CertRotation,
 	)
-	if results.WithResults(res).HasError() {
+	results.WithResults(res)
+	if res != nil && res.HasError() {
 		return results
 	}
 
-	controllerUser, err := user.ReconcileUsersAndRoles(ctx, d.Client, d.ES, d.DynamicWatches(), d.Recorder())
-	if err != nil {
-		return results.WithError(err)
-	}
-
-	// Patch the Pods to add the expected node labels as annotations. Record the error, if any, but do not stop the
-	// reconciliation loop as we don't want to prevent other updates from being applied to the cluster.
-	results.WithResults(annotatePodsWithNodeLabels(ctx, d.Client, d.ES))
-
-	resourcesState, err := reconcile.NewResourcesStateFromAPI(d.Client, d.ES)
-	if err != nil {
-		return results.WithError(err)
-	}
+	// start the ES observer
 	min, err := version.MinInPods(resourcesState.CurrentPods, label.VersionLabelName)
 	if err != nil {
 		return results.WithError(err)
 	}
 	if min == nil {
 		min = &d.Version
 	}
-
-	warnUnsupportedDistro(resourcesState.AllPods, d.ReconcileState.Recorder)
-
 	observedState := d.Observers.ObservedStateResolver(
 		d.ES,
 		d.newElasticsearchClient(
 			resourcesState,
 			controllerUser,
 			*min,
-			certificateResources.TrustedHTTPCertificates,
+			trustedHTTPCertificates,
 		),
 	)
 
-	// always update the elasticsearch state bits
+	// always update the Elasticsearch state bits with the latest observed state
 	d.ReconcileState.UpdateElasticsearchState(*resourcesState, observedState())
 
+	_, res = certificates.ReconcileTransport(
+		ctx,
+		d,
+		d.ES,
+		d.OperatorParameters.CACertRotation,
+		d.OperatorParameters.CertRotation,
+	)
+	results.WithResults(res)
+	if res != nil && res.HasError() {
+		return results
+	}
+
+	// Patch the Pods to add the expected node labels as annotations. Record the error, if any, but do not stop the
+	// reconciliation loop as we don't want to prevent other updates from being applied to the cluster.
+	results.WithResults(annotatePodsWithNodeLabels(ctx, d.Client, d.ES))
+
 	allowDownscales := d.ES.IsConfiguredToAllowDowngrades()
 	if err := d.verifySupportsExistingPods(resourcesState.CurrentPods); err != nil {
 		if !allowDownscales {
@@ -201,7 +219,7 @@ func (d *defaultDriver) Reconcile(ctx context.Context) *reconciler.Results {
 		resourcesState,
 		controllerUser,
 		*min,
-		certificateResources.TrustedHTTPCertificates,
+		trustedHTTPCertificates,
 	)
 	defer esClient.Close()
 
@@ -306,9 +324,8 @@ func (d *defaultDriver) Reconcile(ctx context.Context) *reconciler.Results {
 
 	// reconcile StatefulSets and nodes configuration
 	res = d.reconcileNodeSpecs(ctx, esReachable, esClient, d.ReconcileState, observedState(), *resourcesState, keystoreResources)
-	results = results.WithResults(res)
-
-	if res.HasError() {
+	results.WithResults(res)
+	if res != nil && res.HasError() {
 		return results
 	}
 

pkg/controller/elasticsearch/reconcile/state.go

@@ -13,6 +13,7 @@ import (
 	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/events"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/version"
+	esclient "github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/client"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/hints"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/label"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/observer"
@@ -124,6 +125,12 @@ func (s *State) UpdateElasticsearchReady(
 	return s.updateWithPhase(esv1.ElasticsearchReadyPhase, resourcesState, observedState)
 }
 
+// UpdateElasticsearchUnknown marks Elasticsearch as being the applying changes phase with an unknown health in the resource status.
+func (s *State) UpdateElasticsearchPendingUnknownHealth() *State {
+	unknownState := observer.State{ClusterHealth: &esclient.Health{Status: esv1.ElasticsearchUnknownHealth}}
+	return s.updateWithPhase(esv1.ElasticsearchApplyingChangesPhase, ResourcesState{}, unknownState)
+}
+
 // IsElasticsearchReady reports if Elasticsearch is ready.
 func (s *State) IsElasticsearchReady(observedState observer.State) bool {
 	return s.status.Phase == esv1.ElasticsearchReadyPhase