Merge pull request #2282 from vyncint/feat-fluentbit-flush-on-shutdown

Author: csatib02

charts/logging-operator/charts/logging-operator-crds/templates/logging.banzaicloud.io_fluentbitagents.yaml

@@ -478,6 +478,8 @@ spec:
                 type: object
               bufferStorage:
                 properties:
+                  storage.backlog.flush_on_shutdown:
+                    type: string
                   storage.backlog.mem_limit:
                     type: string
                   storage.checksum:

charts/logging-operator/charts/logging-operator-crds/templates/logging.banzaicloud.io_loggings.yaml

@@ -1339,6 +1339,8 @@ spec:
                     type: object
                   bufferStorage:
                     properties:
+                      storage.backlog.flush_on_shutdown:
+                        type: string
                       storage.backlog.mem_limit:
                         type: string
                       storage.checksum:

charts/logging-operator/crds/logging.banzaicloud.io_fluentbitagents.yaml

@@ -475,6 +475,8 @@ spec:
                 type: object
               bufferStorage:
                 properties:
+                  storage.backlog.flush_on_shutdown:
+                    type: string
                   storage.backlog.mem_limit:
                     type: string
                   storage.checksum:

charts/logging-operator/crds/logging.banzaicloud.io_loggings.yaml

@@ -1336,6 +1336,8 @@ spec:
                     type: object
                   bufferStorage:
                     properties:
+                      storage.backlog.flush_on_shutdown:
+                        type: string
                       storage.backlog.mem_limit:
                         type: string
                       storage.checksum:

config/crd/bases/logging.banzaicloud.io_fluentbitagents.yaml

@@ -475,6 +475,8 @@ spec:
                 type: object
               bufferStorage:
                 properties:
+                  storage.backlog.flush_on_shutdown:
+                    type: string
                   storage.backlog.mem_limit:
                     type: string
                   storage.checksum:

config/crd/bases/logging.banzaicloud.io_loggings.yaml

@@ -1336,6 +1336,8 @@ spec:
                     type: object
                   bufferStorage:
                     properties:
+                      storage.backlog.flush_on_shutdown:
+                        type: string
                       storage.backlog.mem_limit:
                         type: string
                       storage.checksum:

docs/configuration/crds/v1beta1/fluentbit_types.md

@@ -364,6 +364,12 @@ Default: disabled
 
 BufferStorage is the Service Section Configuration of fluent-bit
 
+### storage.backlog.flush_on_shutdown (string, optional) {#bufferstorage-storage.backlog.flush_on_shutdown}
+
+If enabled, Fluent Bit attempts to flush all backlog filesystem chunks to their destination during the shutdown process. This avoids losing buffered records when a node is drained or a spot instance is reclaimed.
+
+Default: Off
+
 ### storage.backlog.mem_limit (string, optional) {#bufferstorage-storage.backlog.mem_limit}
 
 If storage.path is set, Fluent Bit will look for data chunks that were not delivered and are still in the storage layer, these are called backlog data. This option configure a hint of maximum value of memory to use when processing these records.

pkg/resources/fluentbit/configsecret_test.go

@@ -18,8 +18,10 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1"
+	"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/types"
 )
 
 func TestInvalidFilterGrepConfig(t *testing.T) {
@@ -53,3 +55,54 @@ func TestValidFilterGrepConfig(t *testing.T) {
 	assert.NoError(t, err)
 	assert.EqualValues(t, parserFluentdFilterGrep, expectedFluentFilterGrep)
 }
+
+func TestBufferStorageServiceSection(t *testing.T) {
+	testCases := []struct {
+		name          string
+		bufferStorage v1beta1.BufferStorage
+		expected      []string
+		notExpected   []string
+	}{
+		{
+			name: "flush_on_shutdown enabled",
+			bufferStorage: v1beta1.BufferStorage{
+				StoragePath:                   "/buffers",
+				StorageBacklogFlushOnShutdown: "On",
+			},
+			expected: []string{
+				"storage.path  /buffers",
+				"storage.backlog.flush_on_shutdown  On",
+			},
+		},
+		{
+			name: "flush_on_shutdown disabled explicitly",
+			bufferStorage: v1beta1.BufferStorage{
+				StorageBacklogFlushOnShutdown: "Off",
+			},
+			expected: []string{"storage.backlog.flush_on_shutdown  Off"},
+		},
+		{
+			name:          "flush_on_shutdown unset stays out of the config",
+			bufferStorage: v1beta1.BufferStorage{StoragePath: "/buffers"},
+			expected:      []string{"storage.path  /buffers"},
+			notExpected:   []string{"storage.backlog.flush_on_shutdown"},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			mapped, err := types.NewStructToStringMapper(nil).StringsMap(testCase.bufferStorage)
+			require.NoError(t, err)
+
+			rendered, err := generateConfig(fluentBitConfig{BufferStorage: mapped})
+			require.NoError(t, err)
+
+			for _, expected := range testCase.expected {
+				assert.Contains(t, rendered, expected)
+			}
+			for _, notExpected := range testCase.notExpected {
+				assert.NotContains(t, rendered, notExpected)
+			}
+		})
+	}
+}

pkg/sdk/logging/api/v1beta1/fluentbit_types.go

@@ -222,6 +222,8 @@ type BufferStorage struct {
 	StorageDeleteIrrecoverableChunks string `json:"storage.delete_irrecoverable_chunks,omitempty"`
 	// If storage.path is set, Fluent Bit will look for data chunks that were not delivered and are still in the storage layer, these are called backlog data. This option configure a hint of maximum value of memory to use when processing these records. (default:5M)
 	StorageBacklogMemLimit string `json:"storage.backlog.mem_limit,omitempty"`
+	// If enabled, Fluent Bit attempts to flush all backlog filesystem chunks to their destination during the shutdown process. This avoids losing buffered records when a node is drained or a spot instance is reclaimed. (default:Off)
+	StorageBacklogFlushOnShutdown string `json:"storage.backlog.flush_on_shutdown,omitempty"`
 	// Available in Logging operator version 4.4 and later. If the `http_server` option has been enabled in the main Service configuration section, this option registers a new endpoint where internal metrics of the storage layer can be consumed. (default:Off)
 	StorageMetrics string `json:"storage.metrics,omitempty"`
 	// If the input plugin has enabled filesystem storage type, this property sets the maximum number of Chunks that can be up in memory. This is the setting to use to control memory usage when you enable storage.type filesystem. (default: 128)