Merge pull request #1882 from kube-logging/configurable-pause-backport

feat: make fluentbit pause on chunks overlimit configurable
kube-logging · Dec 16, 2024 · 5d2b977 · 5d2b977
2 parents 72257f5 + 5af4684
commit 5d2b977
Show file tree

Hide file tree

Showing 11 changed files with 37 additions and 9 deletions.
diff --git a/charts/logging-operator/crds/logging.banzaicloud.io_fluentbitagents.yaml b/charts/logging-operator/crds/logging.banzaicloud.io_fluentbitagents.yaml
@@ -1493,6 +1493,8 @@ spec:
                     items:
                       type: string
                     type: array
+                  storage.pause_on_chunks_overlimit:
+                    type: string
                   storage.type:
                     type: string
                 type: object

diff --git a/charts/logging-operator/crds/logging.banzaicloud.io_loggings.yaml b/charts/logging-operator/crds/logging.banzaicloud.io_loggings.yaml
@@ -2540,6 +2540,8 @@ spec:
                         items:
                           type: string
                         type: array
+                      storage.pause_on_chunks_overlimit:
+                        type: string
                       storage.type:
                         type: string
                     type: object
@@ -11960,6 +11962,8 @@ spec:
                               items:
                                 type: string
                               type: array
+                            storage.pause_on_chunks_overlimit:
+                              type: string
                             storage.type:
                               type: string
                           type: object

diff --git a/charts/logging-operator/crds/logging.banzaicloud.io_nodeagents.yaml b/charts/logging-operator/crds/logging.banzaicloud.io_nodeagents.yaml
@@ -4055,6 +4055,8 @@ spec:
                         items:
                           type: string
                         type: array
+                      storage.pause_on_chunks_overlimit:
+                        type: string
                       storage.type:
                         type: string
                     type: object

diff --git a/config/crd/bases/logging.banzaicloud.io_fluentbitagents.yaml b/config/crd/bases/logging.banzaicloud.io_fluentbitagents.yaml
@@ -1493,6 +1493,8 @@ spec:
                     items:
                       type: string
                     type: array
+                  storage.pause_on_chunks_overlimit:
+                    type: string
                   storage.type:
                     type: string
                 type: object

diff --git a/config/crd/bases/logging.banzaicloud.io_loggings.yaml b/config/crd/bases/logging.banzaicloud.io_loggings.yaml
@@ -2540,6 +2540,8 @@ spec:
                         items:
                           type: string
                         type: array
+                      storage.pause_on_chunks_overlimit:
+                        type: string
                       storage.type:
                         type: string
                     type: object
@@ -11960,6 +11962,8 @@ spec:
                               items:
                                 type: string
                               type: array
+                            storage.pause_on_chunks_overlimit:
+                              type: string
                             storage.type:
                               type: string
                           type: object

diff --git a/config/crd/bases/logging.banzaicloud.io_nodeagents.yaml b/config/crd/bases/logging.banzaicloud.io_nodeagents.yaml
@@ -4055,6 +4055,8 @@ spec:
                         items:
                           type: string
                         type: array
+                      storage.pause_on_chunks_overlimit:
+                        type: string
                       storage.type:
                         type: string
                     type: object

diff --git a/docs/configuration/crds/v1beta1/fluentbit_types.md b/docs/configuration/crds/v1beta1/fluentbit_types.md
@@ -554,6 +554,12 @@ When a monitored file reach it buffer capacity due to a very long line (Buffer_M
 
 Default: Off
 
+### storage.pause_on_chunks_overlimit (string, optional) {#inputtail-storage.pause_on_chunks_overlimit}
+
+Specifies whether to pause or drop data when the buffer is full.  This helps to make sure we apply backpressure on the input if enabled, see https://docs.fluentbit.io/manual/administration/backpressure
+
+Default: on
+
 ### storage.type (string, optional) {#inputtail-storage.type}
 
 Specify the buffering mechanism to use. It can be memory or filesystem.

diff --git a/pkg/resources/fluentbit/tenants.go b/pkg/resources/fluentbit/tenants.go
@@ -138,8 +138,6 @@ func (r *Reconciler) configureInputsForTenants(tenants []v1beta1.Tenant, input *
 
 		tenantValues["DB"] = fmt.Sprintf("/tail-db/tail-containers-state-%s.db", t.Name)
 		tenantValues["Tag"] = fmt.Sprintf("kubernetes.%s.*", hashFromTenantName(t.Name))
-		// This helps to make sure we apply backpressure on the input, see https://docs.fluentbit.io/manual/administration/backpressure
-		tenantValues["storage.pause_on_chunks_overlimit"] = "on"
 		input.Inputs = append(input.Inputs, fluentbitInputConfigWithTenant{
 			Tenant:          t.Name,
 			Values:          tenantValues,

diff --git a/pkg/resources/nodeagent/nodeagent.go b/pkg/resources/nodeagent/nodeagent.go
@@ -85,12 +85,13 @@ func NodeAgentFluentbitDefaults(userDefined v1beta1.NodeAgentConfig) (*v1beta1.N
 			LogLevel:      "info",
 			CoroStackSize: 24576,
 			InputTail: v1beta1.InputTail{
-				Path:            "/var/log/containers/*.log",
-				RefreshInterval: "5",
-				SkipLongLines:   "On",
-				DB:              util.StringPointer("/tail-db/tail-containers-state.db"),
-				MemBufLimit:     "5MB",
-				Tag:             "kubernetes.*",
+				Path:                          "/var/log/containers/*.log",
+				RefreshInterval:               "5",
+				SkipLongLines:                 "On",
+				DB:                            util.StringPointer("/tail-db/tail-containers-state.db"),
+				MemBufLimit:                   "5MB",
+				Tag:                           "kubernetes.*",
+				StoragePauseOnChunksOverlimit: "on",
 			},
 			Security: &v1beta1.Security{
 				RoleBasedAccessControlCreate: util.BoolPointer(true),
@@ -208,7 +209,8 @@ var NodeAgentFluentbitWindowsDefaults = &v1beta1.NodeAgentConfig{
 			KubeTagPrefix: "kubernetes.C.var.log.containers.",
 		},
 		InputTail: v1beta1.InputTail{
-			Path: "C:\\var\\log\\containers\\*.log",
+			Path:                          "C:\\var\\log\\containers\\*.log",
+			StoragePauseOnChunksOverlimit: "on",
 		},
 		ContainersPath: "C:\\ProgramData\\docker",
 		VarLogsPath:    "C:\\var\\log",

diff --git a/pkg/sdk/logging/api/v1beta1/fluentbit_types.go b/pkg/sdk/logging/api/v1beta1/fluentbit_types.go
@@ -286,6 +286,9 @@ type InputTail struct {
 	DockerModeFlush string `json:"Docker_Mode_Flush,omitempty"`
 	// Specify one or multiple parser definitions to apply to the content. Part of the new Multiline Core support in 1.8 (default: "")
 	MultilineParser []string `json:"multiline.parser,omitempty"`
+	// Specifies whether to pause or drop data when the buffer is full. (default:on)
+	// This helps to make sure we apply backpressure on the input if enabled, see https://docs.fluentbit.io/manual/administration/backpressure
+	StoragePauseOnChunksOverlimit string `json:"storage.pause_on_chunks_overlimit,omitempty"`
 }
 
 // FilterKubernetes Fluent Bit Kubernetes Filter allows to enrich your log files with Kubernetes metadata.

diff --git a/pkg/sdk/logging/api/v1beta1/logging_types.go b/pkg/sdk/logging/api/v1beta1/logging_types.go
@@ -294,6 +294,9 @@ func FluentBitDefaults(fluentbitSpec *FluentbitSpec) error {
 		if fluentbitSpec.InputTail.Tag == "" {
 			fluentbitSpec.InputTail.Tag = "kubernetes.*"
 		}
+		if fluentbitSpec.InputTail.StoragePauseOnChunksOverlimit == "" {
+			fluentbitSpec.InputTail.StoragePauseOnChunksOverlimit = "on"
+		}
 		if fluentbitSpec.Annotations == nil {
 			fluentbitSpec.Annotations = make(map[string]string)
 		}