Support for Logstash secure settings from Kubernetes Secrets using keystore

docs/orchestrating-elastic-stack-applications/logstash.asciidoc

@@ -378,7 +378,7 @@ The Logstash ECK operator creates a user called `eck_logstash_user_role` when an
     }
 
 ```
-You can <<{p}-users-and-roles,update user permissions>> to include more indices if the Elasticsearch plugin is expected to use indices other than the default. See the <<{p}-logstash-configuration-custom-index, Logstash configuration with a custom index>> sample configuration that creates a user that writes to a custom index.
+You can <<{p}-users-and-roles,update user permissions>> to include more indices if the Elasticsearch plugin is expected to use indices other than the default. Check out <<{p}-logstash-configuration-custom-index, Logstash configuration with a custom index>> sample configuration that creates a user that writes to a custom index.
 --
 
 This example demonstrates how to create a Logstash deployment that connects to
@@ -646,6 +646,64 @@ spec:
 ----
 <1> This will change the maximum and minimum heap size of the JVM on each pod to 2GB
 
+[id="{p}-logstash-keystore"]
+=== Setting keystore
+
+You can specify sensitive settings with Kubernetes secrets. ECK automatically injects these settings into the keystore on each Logstash before it starts Logstash.
+The ECK operator continues to watch the secrets for changes and will restart Logstash pods when it detects a change.
+
+NOTE: For the technical preview, the use of settings in the Logstash keystore may impact startup time for Logstash Pods. Startup time will increase linearly for each entry added to the keystore, and this could extend startup time significantly.
+
+The Logstash Keystore can be password protected by setting an environment variable called `LOGSTASH_KEYSTORE_PASS`. Check out https://www.elastic.co/guide/en/logstash/current/keystore.html#keystore-password[Logstash Keystore] documentation for details.
+
+[source,yaml,subs="attributes,+macros,callouts"]
+----
+apiVersion: v1
+kind: Secret
+metadata:
+  name: logstash-keystore-pass
+stringData:
+  LOGSTASH_KEYSTORE_PASS: changed   <1>
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: logstash-secure-settings
+stringData:
+  HELLO: Hallo
+---
+apiVersion: logstash.k8s.elastic.co/v1alpha1
+kind: Logstash
+metadata:
+  name: logstash-sample
+spec:
+  version: 8.8.0
+  count: 1
+  pipelines:
+    - pipeline.id: main
+      config.string: |-
+        input { exec { command => 'uptime' interval => 10 } }
+        filter {
+          if ("${HELLO:}" != "") {   <2>
+            mutate { add_tag => ["awesome"] }
+          }
+        }
+  secureSettings:
+    - secretName: logstash-secure-settings
+  podTemplate:
+    spec:
+      containers:
+        - name: logstash
+          env:
+            - name: LOGSTASH_KEYSTORE_PASS
+              valueFrom:
+                secretKeyRef:
+                  name: logstash-keystore-pass
+                  key: LOGSTASH_KEYSTORE_PASS
+----
+<1> Value of password to protect the Logstash keystore
+<2> The syntax for referencing keys is identical to the syntax for environment variables
+
 [id="{p}-logstash-scaling-logstash"]
 == Scaling Logstash
 

pkg/controller/logstash/driver.go

@@ -6,14 +6,14 @@ package logstash
 
 import (
 	"context"
-
 	"hash/fnv"
 
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
 	"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/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/tracing"
@@ -35,7 +35,8 @@ type Params struct {
 	Logstash logstashv1alpha1.Logstash
 	Status   logstashv1alpha1.LogstashStatus
 
-	OperatorParams operator.Parameters
+	OperatorParams    operator.Parameters
+	KeystoreResources *keystore.Resources
 }
 
 // K8sClient returns the Kubernetes client.
@@ -99,6 +100,12 @@ func internalReconcile(params Params) (*reconciler.Results, logstashv1alpha1.Log
 	params.Logstash.Spec.VolumeClaimTemplates = volume.AppendDefaultPVCs(params.Logstash.Spec.VolumeClaimTemplates,
 		params.Logstash.Spec.PodTemplate.Spec)
 
+	if keystoreResources, err := reconcileKeystore(params, configHash); err != nil {
+		return results.WithError(err), params.Status
+	} else if keystoreResources != nil {
+		params.KeystoreResources = keystoreResources
+	}
+
 	podTemplate, err := buildPodTemplate(params, configHash)
 	if err != nil {
 		return results.WithError(err), params.Status

pkg/controller/logstash/keystore.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 logstash
+
+import (
+	"hash"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+
+	logstashv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/logstash/v1alpha1"
+	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/keystore"
+	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/logstash/volume"
+)
+
+const (
+	KeystorePassKey = "LOGSTASH_KEYSTORE_PASS" // #nosec G101
+)
+
+var (
+	keystoreCommand          = "echo 'y' | /usr/share/logstash/bin/logstash-keystore"
+	initContainersParameters = keystore.InitContainerParameters{
+		KeystoreCreateCommand:         keystoreCommand + " create",
+		KeystoreAddCommand:            keystoreCommand + ` add "$key" --stdin < "$filename"`,
+		SecureSettingsVolumeMountPath: keystore.SecureSettingsVolumeMountPath,
+		KeystoreVolumePath:            volume.ConfigMountPath,
+		Resources: corev1.ResourceRequirements{
+			Requests: map[corev1.ResourceName]resource.Quantity{
+				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourceCPU:    resource.MustParse("1000m"),
+			},
+			Limits: map[corev1.ResourceName]resource.Quantity{
+				corev1.ResourceMemory: resource.MustParse("1Gi"),
+				corev1.ResourceCPU:    resource.MustParse("1000m"),
+			},
+		},
+	}
+)
+
+func reconcileKeystore(params Params, configHash hash.Hash) (*keystore.Resources, error) {
+	if keystoreResources, err := keystore.ReconcileResources(
+		params.Context,
+		params,
+		&params.Logstash,
+		logstashv1alpha1.Namer,
+		NewLabels(params.Logstash),
+		initContainersParameters,
+	); err != nil {
+		return nil, err
+	} else if keystoreResources != nil {
+		_, _ = configHash.Write([]byte(keystoreResources.Version))
+		// set keystore password in init container
+		if env := getKeystorePass(params.Logstash); env != nil {
+			keystoreResources.InitContainer.Env = append(keystoreResources.InitContainer.Env, *env)
+		}
+
+		return keystoreResources, nil
+	}
+
+	return nil, nil
+}
+
+// getKeystorePass return env LOGSTASH_KEYSTORE_PASS from main container if set
+func getKeystorePass(logstash logstashv1alpha1.Logstash) *corev1.EnvVar {
+	for _, c := range logstash.Spec.PodTemplate.Spec.Containers {
+		if c.Name == logstashv1alpha1.LogstashContainerName {
+			for _, env := range c.Env {
+				if env.Name == KeystorePassKey {
+					return &env
+				}
+			}
+		}
+	}
+
+	return nil
+}

pkg/controller/logstash/pod.go

@@ -81,6 +81,12 @@ func buildPodTemplate(params Params, configHash hash.Hash32) (corev1.PodTemplate
 
 	ports := getDefaultContainerPorts()
 
+	if params.KeystoreResources != nil {
+		builder = builder.
+			WithVolumes(params.KeystoreResources.Volume).
+			WithInitContainers(params.KeystoreResources.InitContainer)
+	}
+
 	builder = builder.
 		WithResources(DefaultResources).
 		WithLabels(labels).

test/e2e/logstash/keystore_test.go

@@ -0,0 +1,176 @@
+// 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.
+
+//go:build logstash || e2e
+
+package logstash
+
+import (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	commonv1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/common/v1"
+	lsctrl "github.com/elastic/cloud-on-k8s/v2/pkg/controller/logstash"
+	"github.com/elastic/cloud-on-k8s/v2/test/e2e/test"
+	"github.com/elastic/cloud-on-k8s/v2/test/e2e/test/logstash"
+)
+
+var (
+	pipelineConfig = commonv1.Config{
+		Data: map[string]interface{}{
+			"pipeline.id": "main",
+			"config.string": `
+input { generator { count => 1 } } 
+filter {
+  if ("${HELLO:}" != "") {
+    mutate { add_tag => ["ok"] }
+  }
+}
+`,
+		},
+	}
+
+	request = logstash.Request{
+		Name: "pipeline [main]",
+		Path: "/_node/stats/pipelines/main",
+	}
+
+	want = logstash.Want{
+		Match: map[string]string{
+			"pipelines.main.plugins.filters.0.events.out": "1",
+		},
+	}
+)
+
+// TestLogstashKeystoreWithoutPassword Logstash should resolve ${VAR} in pipelines.yml using keystore key value
+func TestLogstashKeystoreWithoutPassword(t *testing.T) {
+	secretName := "ls-keystore-secure-settings"
+
+	secureSecret := corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      secretName,
+			Namespace: test.Ctx().ManagedNamespace(0),
+		},
+		StringData: map[string]string{
+			"HELLO": "HALLO",
+		},
+	}
+
+	before := test.StepsFunc(func(k *test.K8sClient) test.StepList {
+		return test.StepList{}.WithStep(test.Step{
+			Name: "Create secret for keystore",
+			Test: test.Eventually(func() error {
+				return k.CreateOrUpdateSecrets(secureSecret)
+			}),
+		})
+	})
+
+	b := logstash.NewBuilder("test-keystore-with-default-pw").
+		WithNodeCount(1).
+		WithSecureSettings(commonv1.SecretSource{SecretName: secretName}).
+		WithPipelines([]commonv1.Config{pipelineConfig})
+
+	steps := test.StepsFunc(func(k *test.K8sClient) test.StepList {
+		return test.StepList{
+			b.CheckMetricsRequest(k, request, want),
+			test.Step{
+				Name: "Delete secure secret",
+				Test: test.Eventually(func() error {
+					return k.DeleteSecrets(secureSecret)
+				}),
+			},
+		}
+	})
+
+	test.Sequence(before, steps, b).RunSequential(t)
+}
+
+// TestLogstashKeystoreWithPassword Logstash with customized keystore password
+// should resolve ${VAR} in pipelines.yml using keystore key value
+func TestLogstashKeystoreWithPassword(t *testing.T) {
+	secureSettingSecretName := "ls-keystore-pw-secure-settings"
+
+	secureSecret := corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      secureSettingSecretName,
+			Namespace: test.Ctx().ManagedNamespace(0),
+		},
+		StringData: map[string]string{
+			"HELLO": "HALLO",
+		},
+	}
+
+	passwordSecretName := "ls-keystore-pw"
+
+	passwordSecret := corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      passwordSecretName,
+			Namespace: test.Ctx().ManagedNamespace(0),
+		},
+		StringData: map[string]string{
+			lsctrl.KeystorePassKey: "changed",
+		},
+	}
+
+	before := test.StepsFunc(func(k *test.K8sClient) test.StepList {
+		return test.StepList{}.WithStep(test.Step{
+			Name: "Create secret for keystore",
+			Test: test.Eventually(func() error {
+				return k.CreateOrUpdateSecrets(secureSecret)
+			}),
+		}).WithStep(test.Step{
+			Name: "Create secret for keystore password",
+			Test: test.Eventually(func() error {
+				return k.CreateOrUpdateSecrets(passwordSecret)
+			}),
+		})
+	})
+
+	b := logstash.NewBuilder("test-keystore-with-default-pw").
+		WithNodeCount(1).
+		WithPipelines([]commonv1.Config{pipelineConfig}).
+		WithSecureSettings(commonv1.SecretSource{SecretName: secureSettingSecretName}).
+		WithPodTemplate(corev1.PodTemplateSpec{
+			Spec: corev1.PodSpec{
+				Containers: []corev1.Container{
+					{
+						Name: "logstash",
+						Env: []corev1.EnvVar{
+							{
+								Name: lsctrl.KeystorePassKey,
+								ValueFrom: &corev1.EnvVarSource{
+									SecretKeyRef: &corev1.SecretKeySelector{
+										LocalObjectReference: corev1.LocalObjectReference{Name: passwordSecretName},
+										Key:                  lsctrl.KeystorePassKey,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		})
+
+	steps := test.StepsFunc(func(k *test.K8sClient) test.StepList {
+		return test.StepList{
+			b.CheckMetricsRequest(k, request, want),
+			test.Step{
+				Name: "Delete secure secret",
+				Test: test.Eventually(func() error {
+					return k.DeleteSecrets(secureSecret)
+				}),
+			},
+			test.Step{
+				Name: "Delete keystore pw secret",
+				Test: test.Eventually(func() error {
+					return k.DeleteSecrets(passwordSecret)
+				}),
+			},
+		}
+	})
+
+	test.Sequence(before, steps, b).RunSequential(t)
+}

test/e2e/test/logstash/builder.go

@@ -190,6 +190,16 @@ func (b Builder) WithConfig(config map[string]interface{}) Builder {
 	return b
 }
 
+func (b Builder) WithSecureSettings(secretSource ...commonv1.SecretSource) Builder {
+	b.Logstash.Spec.SecureSettings = append(b.Logstash.Spec.SecureSettings, secretSource...)
+	return b
+}
+
+func (b Builder) WithPodTemplate(podTemplate corev1.PodTemplateSpec) Builder {
+	b.Logstash.Spec.PodTemplate = podTemplate
+	return b
+}
+
 func (b Builder) Name() string {
 	return b.Logstash.Name
 }