Support for Logstash secure settings from Kubernetes Secrets using keystore

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

@@ -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.
+
+You can set keystore password by passing an environment variable called `LOGSTASH_KEYSTORE_PASS`.
+
+[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> The keystore password you set.
+<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 sets
+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
 }