Add TLS and basic authentication integration to Logstash API server

pkg/apis/logstash/v1alpha1/logstash_types.go

@@ -386,6 +386,21 @@ func (l *Logstash) MonitoringAssociation(esRef commonv1.ObjectSelector) commonv1
 	}
 }
 
+// APIServerService returns the user defined API Service
+func (l *Logstash) APIServerService() LogstashService {
+	for _, service := range l.Spec.Services {
+		if UserServiceName(l.Name, service.Name) == APIServiceName(l.Name) {
+			return service
+		}
+	}
+	return LogstashService{}
+}
+
+// APIServerTLSOptions returns the user defined TLSOptions of API Service
+func (l *Logstash) APIServerTLSOptions() commonv1.TLSOptions {
+	return l.APIServerService().TLS
+}
+
 func init() {
 	SchemeBuilder.Register(&Logstash{}, &LogstashList{})
 }

pkg/apis/logstash/v1alpha1/name_test.go

@@ -78,4 +78,4 @@ func TestLogstashName(t *testing.T) {
 			}
 		})
 	}
-}
+}

pkg/controller/common/pod/spec.go

@@ -17,3 +17,12 @@ func ContainerByName(podSpec corev1.PodSpec, name string) *corev1.Container {
 	}
 	return nil
 }
+
+func InitContainerByName(podSpec corev1.PodSpec, name string) *corev1.Container {
+	for i, c := range podSpec.InitContainers {
+		if c.Name == name {
+			return &podSpec.InitContainers[i]
+		}
+	}
+	return nil
+}

pkg/controller/logstash/config.go

@@ -5,7 +5,9 @@
 package logstash
 
 import (
+	"fmt"
 	"hash"
+	"strconv"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -16,18 +18,28 @@ import (
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/reconciler"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/settings"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/tracing"
+	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/logstash/volume"
 )
 
 const (
-	ConfigFileName = "logstash.yml"
+	ConfigFileName         = "logstash.yml"
+	APIKeystorePath        = volume.ConfigMountPath + "/" + APIKeystoreFileName
+	APIKeystoreFileName    = "api_keystore.p12"  // #nosec G101
+	APIKeystoreDefaultPass = "ch@ng3m3"          // #nosec G101
+	APIKeystorePassEnv     = "API_KEYSTORE_PASS" // #nosec G101
 )
 
-func reconcileConfig(params Params, configHash hash.Hash) error {
+func reconcileConfig(params Params, configHash hash.Hash) (*settings.CanonicalConfig, error) {
 	defer tracing.Span(&params.Context)()
 
-	cfgBytes, err := buildConfig(params)
+	cfg, err := buildConfig(params)
 	if err != nil {
-		return err
+		return nil, err
+	}
+
+	cfgBytes, err := cfg.Render()
+	if err != nil {
+		return nil, err
 	}
 
 	expected := corev1.Secret{
@@ -42,28 +54,33 @@ func reconcileConfig(params Params, configHash hash.Hash) error {
 	}
 
 	if _, err = reconciler.ReconcileSecret(params.Context, params.Client, expected, &params.Logstash); err != nil {
-		return err
+		return nil, err
 	}
 
 	_, _ = configHash.Write(cfgBytes)
 
-	return nil
+	return cfg, nil
 }
 
-func buildConfig(params Params) ([]byte, error) {
+func buildConfig(params Params) (*settings.CanonicalConfig, error) {
 	userProvidedCfg, err := getUserConfig(params)
 	if err != nil {
 		return nil, err
 	}
 
 	cfg := defaultConfig()
+	tls := tlsConfig(params.UseTLS)
 
 	// merge with user settings last so they take precedence
-	if err := cfg.MergeWith(userProvidedCfg); err != nil {
+	if err := cfg.MergeWith(tls, userProvidedCfg); err != nil {
 		return nil, err
 	}
 
-	return cfg.Render()
+	if err = checkTLSConfig(cfg, params.UseTLS); err != nil {
+		return nil, err
+	}
+
+	return cfg, nil
 }
 
 // getUserConfig extracts the config either from the spec `config` field or from the Secret referenced by spec
@@ -85,3 +102,28 @@ func defaultConfig() *settings.CanonicalConfig {
 
 	return settings.MustCanonicalConfig(settingsMap)
 }
+
+func tlsConfig(useTLS bool) *settings.CanonicalConfig {
+	if !useTLS {
+		return nil
+	}
+	return settings.MustCanonicalConfig(map[string]interface{}{
+		"api.ssl.enabled":           true,
+		"api.ssl.keystore.path":     APIKeystorePath,
+		"api.ssl.keystore.password": APIKeystoreDefaultPass,
+	})
+}
+
+// checkTLSConfig ensures logstash config `api.ssl.enabled` matches the TLS setting of API service, otherwise throws error
+func checkTLSConfig(cfg *settings.CanonicalConfig, useTLS bool) error {
+	sslEnabled, err := cfg.String("api.ssl.enabled")
+	if err != nil {
+		sslEnabled = "false"
+	}
+
+	if strconv.FormatBool(useTLS) != sslEnabled {
+		return fmt.Errorf("API Service `spec.services.tls.selfSignedCertificate.disabled` is set to `%t`, but logstash config `api.ssl.enabled` is set to `%s`", !useTLS, sslEnabled)
+	}
+
+	return nil
+}

pkg/controller/logstash/config_test.go

@@ -28,6 +28,7 @@ func Test_newConfig(t *testing.T) {
 	tests := []struct {
 		name    string
 		args    args
+		useTLS  bool
 		want    string
 		wantErr bool
 	}{
@@ -43,6 +44,27 @@ func Test_newConfig(t *testing.T) {
 config:
     reload:
         automatic: true
+`,
+			wantErr: false,
+		},
+		{
+			name: "no user config with TLS",
+			args: args{
+				runtimeObjs: nil,
+				logstash:    v1alpha1.Logstash{},
+			},
+			useTLS: true,
+			want: `api:
+    http:
+        host: 0.0.0.0
+    ssl:
+        enabled: true
+        keystore:
+            password: ch@ng3m3
+            path: /usr/share/logstash/config/api_keystore.p12
+config:
+    reload:
+        automatic: true
 `,
 			wantErr: false,
 		},
@@ -52,13 +74,20 @@ config:
 				runtimeObjs: nil,
 				logstash: v1alpha1.Logstash{
 					Spec: v1alpha1.LogstashSpec{Config: &commonv1.Config{Data: map[string]interface{}{
-						"log.level": "debug",
+						"log.level":                 "debug",
+						"api.ssl.keystore.password": "Str0ngP@ssw0rd",
 					}}},
 				},
 			},
+			useTLS: true,
 			want: `api:
     http:
         host: 0.0.0.0
+    ssl:
+        enabled: true
+        keystore:
+            password: Str0ngP@ssw0rd
+            path: /usr/share/logstash/config/api_keystore.p12
 config:
     reload:
         automatic: true
@@ -110,6 +139,29 @@ log:
 			},
 			wantErr: true,
 		},
+		{
+			name: "logstash config disables TLS and service enables TLS",
+			args: args{
+				runtimeObjs: nil,
+				logstash: v1alpha1.Logstash{
+					Spec: v1alpha1.LogstashSpec{
+						Config: &commonv1.Config{Data: map[string]interface{}{
+							"api.ssl.enabled": "false",
+						}},
+						Services: []v1alpha1.LogstashService{{
+							Name: LogstashAPIServiceName,
+							TLS: commonv1.TLSOptions{
+								SelfSignedCertificate: &commonv1.SelfSignedCertificate{
+									Disabled: false,
+								},
+							},
+						}},
+					},
+				},
+			},
+			useTLS:  true,
+			wantErr: true,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -119,19 +171,24 @@ log:
 				EventRecorder: record.NewFakeRecorder(10),
 				Watches:       watches.NewDynamicWatches(),
 				Logstash:      tt.args.logstash,
+				UseTLS:        tt.useTLS,
 			}
 
 			got, err := buildConfig(params)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("newConfig() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
+
 			if tt.wantErr {
 				return // no point in checking the config contents
 			}
 			require.NoError(t, err)
-			if string(got) != tt.want {
-				t.Errorf("newConfig() got = \n%v\n, want \n%v\n", string(got), tt.want)
+
+			gotBytes, _ := got.Render()
+			gotStr := string(gotBytes)
+			if gotStr != tt.want {
+				t.Errorf("newConfig() got = \n%v\n, want \n%v\n", gotStr, tt.want)
 			}
 		})
 	}

pkg/controller/logstash/driver.go

@@ -13,9 +13,12 @@ import (
 	"k8s.io/client-go/tools/record"
 
 	logstashv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/logstash/v1alpha1"
+	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/certificates"
+	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/events"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/keystore"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/operator"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/reconciler"
+	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/settings"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/tracing"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/watches"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/logstash/stackmon"
@@ -37,6 +40,8 @@ type Params struct {
 
 	OperatorParams    operator.Parameters
 	KeystoreResources *keystore.Resources
+	UseTLS            bool                      // Logstash API Server uses TLS
+	LogstashConfig    *settings.CanonicalConfig // logstash.yml config
 }
 
 // K8sClient returns the Kubernetes client.
@@ -74,19 +79,44 @@ func internalReconcile(params Params) (*reconciler.Results, logstashv1alpha1.Log
 	defer tracing.Span(&params.Context)()
 	results := reconciler.NewResult(params.Context)
 
-	_, err := reconcileServices(params)
+	_, apiSvc, err := reconcileServices(params)
 	if err != nil {
 		return results.WithError(err), params.Status
 	}
 
+	apiSvcTLS := params.Logstash.APIServerTLSOptions()
+
+	_, results = certificates.Reconciler{
+		K8sClient:             params.Client,
+		DynamicWatches:        params.Watches,
+		Owner:                 &params.Logstash,
+		TLSOptions:            apiSvcTLS,
+		Namer:                 logstashv1alpha1.Namer,
+		Labels:                NewLabels(params.Logstash),
+		Services:              []corev1.Service{apiSvc},
+		GlobalCA:              params.OperatorParams.GlobalCA,
+		CACertRotation:        params.OperatorParams.CACertRotation,
+		CertRotation:          params.OperatorParams.CertRotation,
+		GarbageCollectSecrets: true,
+	}.ReconcileCAAndHTTPCerts(params.Context)
+	if results.HasError() {
+		_, err := results.Aggregate()
+		k8s.MaybeEmitErrorEvent(params.Recorder(), err, &params.Logstash, events.EventReconciliationError, "Certificate reconciliation error: %v", err)
+		return results, params.Status
+	}
+
+	params.UseTLS = apiSvcTLS.Enabled()
+
 	configHash := fnv.New32a()
 
-	// reconcile beats config secrets if Stack Monitoring is defined
-	if err := stackmon.ReconcileConfigSecrets(params.Context, params.Client, params.Logstash); err != nil {
+	cfg, err := reconcileConfig(params, configHash)
+	if err != nil {
 		return results.WithError(err), params.Status
 	}
+	params.LogstashConfig = cfg
 
-	if err := reconcileConfig(params, configHash); err != nil {
+	// reconcile beats config secrets if Stack Monitoring is defined
+	if err := stackmon.ReconcileConfigSecrets(params.Context, params.Client, params.Logstash, params.UseTLS, params.LogstashConfig); err != nil {
 		return results.WithError(err), params.Status
 	}