Logstash - add ability to reload pipeline(s) without triggering full pod restart

config/crds/v1/bases/logstash.k8s.elastic.co_logstashes.yaml

@@ -1944,7 +1944,8 @@ spec:
                                     defined in spec.resourceClaims, that are used
                                     by this container. \n This is an alpha field and
                                     requires enabling the DynamicResourceAllocation
-                                    feature gate. \n This field is immutable."
+                                    feature gate. \n This field is immutable. It can
+                                    only be set for containers."
                                   items:
                                     description: ResourceClaim references one entry
                                       in PodSpec.ResourceClaims.
@@ -3314,7 +3315,8 @@ spec:
                                     defined in spec.resourceClaims, that are used
                                     by this container. \n This is an alpha field and
                                     requires enabling the DynamicResourceAllocation
-                                    feature gate. \n This field is immutable."
+                                    feature gate. \n This field is immutable. It can
+                                    only be set for containers."
                                   items:
                                     description: ResourceClaim references one entry
                                       in PodSpec.ResourceClaims.
@@ -4717,7 +4719,8 @@ spec:
                                     defined in spec.resourceClaims, that are used
                                     by this container. \n This is an alpha field and
                                     requires enabling the DynamicResourceAllocation
-                                    feature gate. \n This field is immutable."
+                                    feature gate. \n This field is immutable. It can
+                                    only be set for containers."
                                   items:
                                     description: ResourceClaim references one entry
                                       in PodSpec.ResourceClaims.
@@ -6523,7 +6526,8 @@ spec:
                                                 that are used by this container. \n
                                                 This is an alpha field and requires
                                                 enabling the DynamicResourceAllocation
-                                                feature gate. \n This field is immutable."
+                                                feature gate. \n This field is immutable.
+                                                It can only be set for containers."
                                               items:
                                                 description: ResourceClaim references
                                                   one entry in PodSpec.ResourceClaims.

config/samples/logstash/logstash_svc.yaml

@@ -22,6 +22,10 @@ spec:
     api.http.host: "0.0.0.0"
     api.http.port: 9601
     queue.type: memory
+  pipelines:
+    - pipeline.id: one
+      pipeline.workers: 2
+      config.string: "input { beats { port => 5044 }} output { stdout {}}"
   services:
     - name: api
       service:

pkg/controller/logstash/config.go

@@ -75,6 +75,8 @@ func defaultConfig() *settings.CanonicalConfig {
 	settingsMap := map[string]interface{}{
 		// Set 'api.http.host' by default to `0.0.0.0` for readiness probe to work.
 		"api.http.host": "0.0.0.0",
+		// Set `config.reload.automatic` to `true` to enable pipeline reloads by default
+		"config.reload.automatic": true,
 	}
 
 	return settings.MustCanonicalConfig(settingsMap)

pkg/controller/logstash/config_test.go

@@ -40,6 +40,9 @@ func Test_newConfig(t *testing.T) {
 			want: `api:
     http:
         host: 0.0.0.0
+config:
+    reload:
+        automatic: true
 `,
 			wantErr: false,
 		},
@@ -56,6 +59,9 @@ func Test_newConfig(t *testing.T) {
 			want: `api:
     http:
         host: 0.0.0.0
+config:
+    reload:
+        automatic: true
 log:
     level: debug
 `,
@@ -70,6 +76,9 @@ log:
 			want: `api:
     http:
         host: 0.0.0.0
+config:
+    reload:
+        automatic: true
 log:
     level: debug
 `,
@@ -86,6 +95,9 @@ log:
 			want: `api:
     http:
         host: 0.0.0.0
+config:
+    reload:
+        automatic: true
 log:
     level: warn
 `,

pkg/controller/logstash/driver.go

@@ -88,7 +88,10 @@ func internalReconcile(params Params) (*reconciler.Results, logstashv1alpha1.Log
 		return results.WithError(err), params.Status
 	}
 
-	if err := reconcilePipeline(params, configHash); err != nil {
+	// We intentionally DO NOT pass the configHash here. We don't want to consider the pipeline definitions in the
+	// hash of the config to ensure that a pipeline change does not automatically trigger a restart
+	// of the pod, but allows Logstash's automatic reload of pipelines to take place
+	if err := reconcilePipeline(params); err != nil {
 		return results.WithError(err), params.Status
 	}
 

pkg/controller/logstash/initcontainer.go

@@ -0,0 +1,80 @@
+// 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 (
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+
+	logstashv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/logstash/v1alpha1"
+)
+
+const (
+	InitConfigContainerName = "logstash-internal-init-config"
+
+	// InitConfigScript is a small bash script to prepare the logstash configuration directory
+	InitConfigScript = `#!/usr/bin/env bash
+set -eu
+
+init_config_initialized_flag=` + InitContainerConfigVolumeMountPath + `/elastic-internal-init-config.ok
+
+if [[ -f "${init_config_initialized_flag}" ]]; then
+    echo "Logstash configuration already initialized."
+	exit 0
+fi
+
+echo "Setup Logstash configuration"
+
+mount_path=` + InitContainerConfigVolumeMountPath + `
+
+for f in /usr/share/logstash/config/*.*; do
+	filename=$(basename $f)
+	if [[ ! -f "$mount_path/$filename" ]]; then
+		cp $f $mount_path
+	fi
+done
+
+ln -sf ` + InternalConfigVolumeMountPath + `/logstash.yml  $mount_path
+ln -sf ` + InternalPipelineVolumeMountPath + `/pipelines.yml  $mount_path
+
+touch "${init_config_initialized_flag}"
+echo "Logstash configuration successfully prepared."
+`
+)
+
+// initConfigContainer returns an init container that executes a bash script to prepare the logstash config directory.
+// This copies files from the `config` folder of the docker image, and creates symlinks for the operator created
+// `logstash.yml` and `pipelines.yml` file into a shared config folder to use by the main logstash container. This
+// enables dynamic reloads for `pipelines.yml`
+func initConfigContainer(ls logstashv1alpha1.Logstash) corev1.Container {
+	privileged := false
+
+	return corev1.Container{
+		// Image will be inherited from pod template defaults
+		ImagePullPolicy: corev1.PullIfNotPresent,
+		Name:            InitConfigContainerName,
+		SecurityContext: &corev1.SecurityContext{
+			Privileged: &privileged,
+		},
+		Command: []string{"/usr/bin/env", "bash", "-c", InitConfigScript},
+		VolumeMounts: []corev1.VolumeMount{
+			ConfigSharedVolume.InitContainerVolumeMount(),
+			ConfigVolume(ls).VolumeMount(),
+			PipelineVolume(ls).VolumeMount(),
+		},
+
+		Resources: corev1.ResourceRequirements{
+			Requests: map[corev1.ResourceName]resource.Quantity{
+				corev1.ResourceMemory: resource.MustParse("50Mi"),
+				corev1.ResourceCPU:    resource.MustParse("0.1"),
+			},
+			Limits: map[corev1.ResourceName]resource.Quantity{
+				// Memory limit should be at least 12582912 when running with CRI-O
+				corev1.ResourceMemory: resource.MustParse("50Mi"),
+				corev1.ResourceCPU:    resource.MustParse("0.1"),
+			},
+		},
+	}
+}

pkg/controller/logstash/labels.go

@@ -5,6 +5,8 @@
 package logstash
 
 import (
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
 	commonv1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/common/v1"
 	logstashv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/logstash/v1alpha1"
 )
@@ -27,3 +29,8 @@ func NewLabels(logstash logstashv1alpha1.Logstash) map[string]string {
 		NameLabelName:          logstash.Name,
 	}
 }
+
+// NewLabelSelectorForLogstash returns a labels.Selector that matches the labels as constructed by NewLabels
+func NewLabelSelectorForLogstash(ls logstashv1alpha1.Logstash) client.MatchingLabels {
+	return client.MatchingLabels(map[string]string{commonv1.TypeLabelName: TypeLabelValue, NameLabelName: ls.Name})
+}

pkg/controller/logstash/pipeline.go

@@ -5,19 +5,24 @@
 package logstash
 
 import (
-	"hash"
+	"reflect"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
 	logstashv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/logstash/v1alpha1"
+	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/annotation"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/labels"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/reconciler"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/tracing"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/logstash/pipelines"
+
+	"github.com/elastic/cloud-on-k8s/v2/pkg/utils/maps"
 )
 
-func reconcilePipeline(params Params, configHash hash.Hash) error {
+func reconcilePipeline(params Params) error {
 	defer tracing.Span(&params.Context)()
 
 	cfgBytes, err := buildPipeline(params)
@@ -36,15 +41,46 @@ func reconcilePipeline(params Params, configHash hash.Hash) error {
 		},
 	}
 
-	if _, err = reconciler.ReconcileSecret(params.Context, params.Client, expected, &params.Logstash); err != nil {
+	if err := reconcileSecretWithFastUpdate(params, expected); err != nil {
 		return err
 	}
-
-	_, _ = configHash.Write(cfgBytes)
-
 	return nil
 }
 
+// This function reconciles the secret, but then adds a postUpdate step to mark the pods as updated
+// to trigger a quicker reload of the updated secret than waiting for the kubelet sync interval to kick in
+func reconcileSecretWithFastUpdate(params Params, expected corev1.Secret) error {
+	var reconciled corev1.Secret
+
+	return reconciler.ReconcileResource(reconciler.Params{
+		Context:    params.Context,
+		Client:     params.Client,
+		Owner:      &params.Logstash,
+		Expected:   &expected,
+		Reconciled: &reconciled,
+		NeedsUpdate: func() bool {
+			// update if expected labels and annotations are not there
+			return !maps.IsSubset(expected.Labels, reconciled.Labels) ||
+				!maps.IsSubset(expected.Annotations, reconciled.Annotations) ||
+				// or if secret data is not strictly equal
+				!reflect.DeepEqual(expected.Data, reconciled.Data)
+		},
+		UpdateReconciled: func() {
+			// set expected annotations and labels, but don't remove existing ones
+			// that may have been defaulted or set by the user on the existing resource
+			reconciled.Labels = maps.Merge(reconciled.Labels, expected.Labels)
+			reconciled.Annotations = maps.Merge(reconciled.Annotations, expected.Annotations)
+			reconciled.Data = expected.Data
+		},
+		PostUpdate: func() {
+			annotation.MarkPodsAsUpdated(params.Context, params.Client,
+				client.InNamespace(params.Logstash.Namespace),
+				NewLabelSelectorForLogstash(params.Logstash),
+			)
+		},
+	})
+}
+
 func buildPipeline(params Params) ([]byte, error) {
 	userProvidedCfg, err := getUserPipeline(params)
 	if err != nil {

pkg/controller/logstash/pod.go

@@ -7,7 +7,6 @@ package logstash
 import (
 	"fmt"
 	"hash"
-	"path"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -39,6 +38,14 @@ const (
 
 	// VersionLabelName is a label used to track the version of a Logstash Pod.
 	VersionLabelName = "logstash.k8s.elastic.co/version"
+
+	InitContainerConfigVolumeMountPath = "/mnt/elastic-internal/logstash-config-local"
+
+	// InternalConfigVolumeName is a volume which contains the generated configuration.
+	InternalConfigVolumeName        = "elastic-internal-logstash-config"
+	InternalConfigVolumeMountPath   = "/mnt/elastic-internal/logstash-config"
+	InternalPipelineVolumeName      = "elastic-internal-logstash-pipeline"
+	InternalPipelineVolumeMountPath = "/mnt/elastic-internal/logstash-pipeline"
 )
 
 var (
@@ -54,26 +61,38 @@ var (
 	}
 )
 
+var (
+	// ConfigSharedVolume contains the Logstash config/ directory, it contains the contents of config from the docker container
+	ConfigSharedVolume = volume.SharedVolume{
+		VolumeName:             ConfigVolumeName,
+		InitContainerMountPath: InitContainerConfigVolumeMountPath,
+		ContainerMountPath:     ConfigMountPath,
+	}
+)
+
+// ConfigVolume returns a SecretVolume to hold the Logstash config of the given Logstash resource.
+func ConfigVolume(ls logstashv1alpha1.Logstash) volume.SecretVolume {
+	return volume.NewSecretVolumeWithMountPath(
+		logstashv1alpha1.ConfigSecretName(ls.Name),
+		InternalConfigVolumeName,
+		InternalConfigVolumeMountPath,
+	)
+}
+
+// PipelineVolume returns a SecretVolume to hold the Logstash config of the given Logstash resource.
+func PipelineVolume(ls logstashv1alpha1.Logstash) volume.SecretVolume {
+	return volume.NewSecretVolumeWithMountPath(
+		logstashv1alpha1.PipelineSecretName(ls.Name),
+		InternalPipelineVolumeName,
+		InternalPipelineVolumeMountPath,
+	)
+}
+
 func buildPodTemplate(params Params, configHash hash.Hash32) corev1.PodTemplateSpec {
 	defer tracing.Span(&params.Context)()
 	spec := &params.Logstash.Spec
 	builder := defaults.NewPodTemplateBuilder(params.GetPodTemplate(), logstashv1alpha1.LogstashContainerName)
-	vols := []volume.VolumeLike{
-		// volume with logstash configuration file
-		volume.NewSecretVolume(
-			logstashv1alpha1.ConfigSecretName(params.Logstash.Name),
-			LogstashConfigVolumeName,
-			path.Join(ConfigMountPath, LogstashConfigFileName),
-			LogstashConfigFileName,
-			0644),
-		// volume with logstash pipeline file
-		volume.NewSecretVolume(
-			logstashv1alpha1.PipelineSecretName(params.Logstash.Name),
-			PipelineVolumeName,
-			path.Join(ConfigMountPath, PipelineFileName),
-			PipelineFileName,
-			0644),
-	}
+	vols := []volume.VolumeLike{ConfigSharedVolume, ConfigVolume(params.Logstash), PipelineVolume(params.Logstash)}
 
 	labels := maps.Merge(params.Logstash.GetIdentityLabels(), map[string]string{
 		VersionLabelName: spec.Version})
@@ -93,6 +112,7 @@ func buildPodTemplate(params Params, configHash hash.Hash32) corev1.PodTemplateS
 		WithPorts(ports).
 		WithReadinessProbe(readinessProbe(params.Logstash)).
 		WithVolumeLikes(vols...).
+		WithInitContainers(initConfigContainer(params.Logstash)).
 		WithInitContainerDefaults()
 
 	builder, err := stackmon.WithMonitoring(params.Context, params.Client, builder, params.Logstash)