Use the public transport CA as remote CA if the remote CA list is emp…

…ty (#3993) (#4007)

* Addresses an issue where remote transport ca.crt is empty if remote clusters are not set for an ES cluster
* Preserves the behavior where it is not needed to restart ES when configuring remote clusters, post the initial deployment of the cluster

pkg/controller/elasticsearch/certificates/reconcile.go

@@ -44,12 +44,7 @@ func Reconcile(
 	span, _ := apm.StartSpan(ctx, "reconcile_certs", tracing.SpanTypeApp)
 	defer span.End()
 
-	results := &reconciler.Results{}
-
-	// reconcile remote clusters certificate authorities
-	if err := remoteca.Reconcile(driver.K8sClient(), es); err != nil {
-		results.WithError(err)
-	}
+	var results *reconciler.Results
 
 	// label certificates secrets with the cluster name
 	certsLabels := label.NewLabels(k8s.ExtractNamespacedName(&es))
@@ -104,6 +99,11 @@ func Reconcile(
 		certRotation,
 	)
 
+	// reconcile remote clusters certificate authorities
+	if err := remoteca.Reconcile(driver.K8sClient(), es, *transportCA); err != nil {
+		results.WithError(err)
+	}
+
 	if results.WithResults(transportResults).HasError() {
 		return nil, results
 	}

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

@@ -36,6 +36,7 @@ func Labels(esName string) client.MatchingLabels {
 func Reconcile(
 	c k8s.Client,
 	es esv1.Elasticsearch,
+	transportCA certificates.CA,
 ) error {
 	// Get all the remote certificate authorities
 	var remoteCAList v1.SecretList
@@ -46,15 +47,21 @@ func Reconcile(
 	); 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]
+	var remoteCertificateAuthorities [][]byte
+	if len(remoteCAList.Items) > 0 {
+		// 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]
+		}
+	} else {
+		// if remoteCAList is empty we use the provided transport CA so that we don't end up having an empty cert file mounted on the ES container
+		remoteCertificateAuthorities = [][]byte{certificates.EncodePEMCert(transportCA.Cert.Raw)}
 	}
 
 	expected := v1.Secret{

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

@@ -22,9 +22,11 @@ import (
 
 func TestReconcile(t *testing.T) {
 	type args struct {
-		es      esv1.Elasticsearch
-		secrets []runtime.Object
+		es          esv1.Elasticsearch
+		secrets     []runtime.Object
+		transportCA certificates.CA
 	}
+	testTransportCA, _ := certificates.NewSelfSignedCA(certificates.CABuilderOptions{})
 	tests := []struct {
 		name    string
 		args    args
@@ -59,6 +61,7 @@ func TestReconcile(t *testing.T) {
 						Data: map[string][]byte{certificates.CAFileName: []byte("cert2\n")},
 					},
 				},
+				transportCA: *testTransportCA,
 			},
 			want: []byte("cert2\ncert1\n"),
 		},
@@ -101,14 +104,24 @@ func TestReconcile(t *testing.T) {
 						Data: map[string][]byte{certificates.CAFileName: []byte("cert2\n")},
 					},
 				},
+				transportCA: *testTransportCA,
 			},
 			want: []byte("cert2\ncert1\n"),
 		},
+		{
+			name: "Use provided transport CA if remote CA list is empty",
+			args: args{
+				es:          esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{Name: "es1", Namespace: "ns1"}},
+				transportCA: *testTransportCA,
+			},
+			want: certificates.EncodePEMCert(testTransportCA.Cert.Raw),
+		},
 	}
+
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			k8sClient := k8s.WrappedFakeClient(tt.args.secrets...)
-			if err := Reconcile(k8sClient, tt.args.es); (err != nil) != tt.wantErr {
+			if err := Reconcile(k8sClient, tt.args.es, tt.args.transportCA); (err != nil) != tt.wantErr {
 				t.Errorf("Reconcile() error = %v, wantErr %v", err, tt.wantErr)
 			}
 			remoteCaList := v1.Secret{}