Merge pull request #1888 from kube-logging/forward-port-release-4.10

forward port release 4.10
kube-logging · Dec 17, 2024 · 0a15baf · 0a15baf
2 parents 096a5a0 + 45110de
commit 0a15baf
Show file tree

Hide file tree

Showing 19 changed files with 125 additions and 38 deletions.
diff --git a/charts/logging-operator/charts/crds/templates/logging.banzaicloud.io_fluentbitagents.yaml b/charts/logging-operator/charts/crds/templates/logging.banzaicloud.io_fluentbitagents.yaml
@@ -1478,6 +1478,8 @@ spec:
               flush:
                 format: int32
                 type: integer
+              forceHotReloadAfterGrace:
+                type: boolean
               forwardOptions:
                 properties:
                   Require_ack_response:
@@ -1585,6 +1587,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/charts/crds/templates/logging.banzaicloud.io_loggings.yaml b/charts/logging-operator/charts/crds/templates/logging.banzaicloud.io_loggings.yaml
@@ -2525,6 +2525,8 @@ spec:
                   flush:
                     format: int32
                     type: integer
+                  forceHotReloadAfterGrace:
+                    type: boolean
                   forwardOptions:
                     properties:
                       Require_ack_response:
@@ -2632,6 +2634,8 @@ spec:
                         items:
                           type: string
                         type: array
+                      storage.pause_on_chunks_overlimit:
+                        type: string
                       storage.type:
                         type: string
                     type: object
@@ -12185,6 +12189,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/charts/crds/templates/logging.banzaicloud.io_nodeagents.yaml b/charts/logging-operator/charts/crds/templates/logging.banzaicloud.io_nodeagents.yaml
@@ -4066,6 +4066,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_fluentbitagents.yaml b/charts/logging-operator/crds/logging.banzaicloud.io_fluentbitagents.yaml
@@ -1475,6 +1475,8 @@ spec:
               flush:
                 format: int32
                 type: integer
+              forceHotReloadAfterGrace:
+                type: boolean
               forwardOptions:
                 properties:
                   Require_ack_response:
@@ -1582,6 +1584,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
@@ -2522,6 +2522,8 @@ spec:
                   flush:
                     format: int32
                     type: integer
+                  forceHotReloadAfterGrace:
+                    type: boolean
                   forwardOptions:
                     properties:
                       Require_ack_response:
@@ -2629,6 +2631,8 @@ spec:
                         items:
                           type: string
                         type: array
+                      storage.pause_on_chunks_overlimit:
+                        type: string
                       storage.type:
                         type: string
                     type: object
@@ -12182,6 +12186,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
@@ -4063,6 +4063,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
@@ -1475,6 +1475,8 @@ spec:
               flush:
                 format: int32
                 type: integer
+              forceHotReloadAfterGrace:
+                type: boolean
               forwardOptions:
                 properties:
                   Require_ack_response:
@@ -1582,6 +1584,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
@@ -2522,6 +2522,8 @@ spec:
                   flush:
                     format: int32
                     type: integer
+                  forceHotReloadAfterGrace:
+                    type: boolean
                   forwardOptions:
                     properties:
                       Require_ack_response:
@@ -2629,6 +2631,8 @@ spec:
                         items:
                           type: string
                         type: array
+                      storage.pause_on_chunks_overlimit:
+                        type: string
                       storage.type:
                         type: string
                     type: object
@@ -12182,6 +12186,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
@@ -4063,6 +4063,8 @@ spec:
                         items:
                           type: string
                         type: array
+                      storage.pause_on_chunks_overlimit:
+                        type: string
                       storage.type:
                         type: string
                     type: object

diff --git a/config/samples/multitenant-routing/logging/tenant-infra-logging.yaml b/config/samples/multitenant-routing/logging/tenant-infra-logging.yaml
@@ -11,7 +11,8 @@ metadata:
     tenant: infra
 spec:
   loggingRef: infra
-  fluentd: {}
+  fluentd:
+    metrics: {}
   controlNamespace: infra
 ---
 apiVersion: logging.banzaicloud.io/v1beta1
@@ -49,8 +50,12 @@ metadata:
   name: infra
 spec:
   loggingRef: infra
+  # this is required to reload even if there are pending tasks in one of the queues
+  # requires grace to be set, which is 5 by default
+  forceHotReloadAfterGrace: true
   inputTail:
     storage.type: filesystem
+    storage.pause_on_chunks_overlimit: "off"
   positiondb:
     hostPath:
       path: ""
@@ -59,9 +64,15 @@ spec:
       path: ""
   network:
     connectTimeout: 2
+    keepaliveMaxRecycle: 20
   metrics: {}
+  bufferStorage:
+    storage.max_chunks_up: 10
+  forwardOptions:
+    storage.total_limit_size: 50MB
   image:
-    tag: 2.2.2-debug
+    tag: 3.1.10-debug
+  configHotReload: {}
 ---
 apiVersion: logging.banzaicloud.io/v1beta1
 kind: LoggingRoute

diff --git a/controllers/logging/logging_controller.go b/controllers/logging/logging_controller.go
@@ -102,6 +102,8 @@ type LoggingReconciler struct {
 func (r *LoggingReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	log := r.Log.WithValues("logging", req.Name)
 
+	log.V(1).Info("reconciling")
+
 	var logging loggingv1beta1.Logging
 	if err := r.Client.Get(ctx, req.NamespacedName, &logging); err != nil {
 		// If object is not found, return without error.

diff --git a/docs/configuration/crds/v1beta1/fluentbit_types.md b/docs/configuration/crds/v1beta1/fluentbit_types.md
@@ -125,6 +125,11 @@ Set the flush time in seconds.nanoseconds. The engine loop uses a Flush timeout
 
 Default: 1
 
+### forceHotReloadAfterGrace (bool, optional) {#fluentbitspec-forcehotreloadaftergrace}
+
+HotReload pauses all inputs and waits until they finish. In certain situations this is unacceptable, for example if an output is down for a longer time. An undocumented option called "Hot_Reload.Ensure_Thread_Safety Off" can be used at the [SERVICE] config to force hotreload after the grace period. Please note that it might result in a SIGSEGV, but worst case kubelet will restart the container. See https://github.com/fluent/fluent-bit/pull/7509 
+
+
 ### forwardOptions (*ForwardOptions, optional) {#fluentbitspec-forwardoptions}
 
 
@@ -557,6 +562,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/main.go b/main.go
@@ -25,6 +25,7 @@ import (
 	"runtime/coverage"
 	"strings"
 	"syscall"
+	"time"
 
 	"emperror.dev/errors"
 	prometheusOperator "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
@@ -81,6 +82,7 @@ func main() {
 	var namespace string
 	var loggingRef string
 	var klogLevel int
+	var syncPeriod string
 
 	flag.StringVar(&metricsAddr, "metrics-addr", ":8080", "The address the metric endpoint binds to.")
 	flag.BoolVar(&enableLeaderElection, "enable-leader-election", false,
@@ -91,6 +93,7 @@ func main() {
 	flag.BoolVar(&enableprofile, "pprof", false, "Enable pprof")
 	flag.StringVar(&namespace, "watch-namespace", "", "Namespace to filter the list of watched objects")
 	flag.StringVar(&loggingRef, "watch-logging-name", "", "Logging resource name to optionally filter the list of watched objects based on which logging they belong to by checking the app.kubernetes.io/managed-by label")
+	flag.StringVar(&syncPeriod, "sync-period", "", "SyncPeriod determines the minimum frequency at which watched resources are reconciled. Defaults to 10 hours. Parsed using time.ParseDuration.")
 	flag.Parse()
 
 	ctx := context.Background()
@@ -142,7 +145,7 @@ func main() {
 		mgrOptions.WebhookServer = webhookServer
 	}
 
-	customMgrOptions, err := setupCustomCache(&mgrOptions, namespace, loggingRef)
+	customMgrOptions, err := setupCustomCache(&mgrOptions, syncPeriod, namespace, loggingRef)
 	if err != nil {
 		setupLog.Error(err, "unable to set up custom cache settings")
 		os.Exit(1)
@@ -285,7 +288,15 @@ func detectContainerRuntime(ctx context.Context, c client.Reader) error {
 	return nil
 }
 
-func setupCustomCache(mgrOptions *ctrl.Options, namespace string, loggingRef string) (*ctrl.Options, error) {
+func setupCustomCache(mgrOptions *ctrl.Options, syncPeriod string, namespace string, loggingRef string) (*ctrl.Options, error) {
+	if syncPeriod != "" {
+		duration, err := time.ParseDuration(syncPeriod)
+		if err != nil {
+			return mgrOptions, err
+		}
+		mgrOptions.Cache.SyncPeriod = &duration
+	}
+
 	if namespace == "" && loggingRef == "" {
 		return mgrOptions, nil
 	}

diff --git a/pkg/resources/fluentbit/config.go b/pkg/resources/fluentbit/config.go
@@ -26,6 +26,9 @@ var fluentBitConfigTemplate = `
 [SERVICE]
     Flush        {{ .Flush }}
     Grace        {{ .Grace }}
+    {{- if .ForceHotReloadAfterGrace }}
+    Hot_Reload.Ensure_Thread_Safety off
+    {{- end }}
     Daemon       Off
     Log_Level    {{ .LogLevel }}
     Parsers_File {{ .DefaultParsers }}

diff --git a/pkg/resources/fluentbit/configsecret.go b/pkg/resources/fluentbit/configsecret.go
@@ -64,23 +64,24 @@ type fluentBitConfig struct {
 		Port    int32
 		Path    string
 	}
-	Flush                   int32
-	Grace                   int32
-	LogLevel                string
-	CoroStackSize           int32
-	Output                  map[string]string
-	Input                   fluentbitInputConfig
-	Inputs                  []fluentbitInputConfigWithTenant
-	DisableKubernetesFilter bool
-	KubernetesFilter        map[string]string
-	AwsFilter               map[string]string
-	BufferStorage           map[string]string
-	FilterModify            []v1beta1.FilterModify
-	FluentForwardOutput     *fluentForwardOutputConfig
-	SyslogNGOutput          *syslogNGOutputConfig
-	DefaultParsers          string
-	CustomParsers           string
-	HealthCheck             *v1beta1.HealthCheck
+	Flush                    int32
+	Grace                    int32
+	LogLevel                 string
+	CoroStackSize            int32
+	Output                   map[string]string
+	ForceHotReloadAfterGrace bool
+	Input                    fluentbitInputConfig
+	Inputs                   []fluentbitInputConfigWithTenant
+	DisableKubernetesFilter  bool
+	KubernetesFilter         map[string]string
+	AwsFilter                map[string]string
+	BufferStorage            map[string]string
+	FilterModify             []v1beta1.FilterModify
+	FluentForwardOutput      *fluentForwardOutputConfig
+	SyslogNGOutput           *syslogNGOutputConfig
+	DefaultParsers           string
+	CustomParsers            string
+	HealthCheck              *v1beta1.HealthCheck
 }
 
 type fluentForwardOutputConfig struct {
@@ -213,14 +214,15 @@ func (r *Reconciler) configSecret() (runtime.Object, reconciler.DesiredState, er
 	}
 
 	input := fluentBitConfig{
-		Flush:                   r.fluentbitSpec.Flush,
-		Grace:                   r.fluentbitSpec.Grace,
-		LogLevel:                r.fluentbitSpec.LogLevel,
-		CoroStackSize:           r.fluentbitSpec.CoroStackSize,
-		Namespace:               r.Logging.Spec.ControlNamespace,
-		DisableKubernetesFilter: disableKubernetesFilter,
-		FilterModify:            r.fluentbitSpec.FilterModify,
-		HealthCheck:             r.fluentbitSpec.HealthCheck,
+		Flush:                    r.fluentbitSpec.Flush,
+		Grace:                    r.fluentbitSpec.Grace,
+		ForceHotReloadAfterGrace: r.fluentbitSpec.ForceHotReloadAfterGrace,
+		LogLevel:                 r.fluentbitSpec.LogLevel,
+		CoroStackSize:            r.fluentbitSpec.CoroStackSize,
+		Namespace:                r.Logging.Spec.ControlNamespace,
+		DisableKubernetesFilter:  disableKubernetesFilter,
+		FilterModify:             r.fluentbitSpec.FilterModify,
+		HealthCheck:              r.fluentbitSpec.HealthCheck,
 	}
 
 	input.DefaultParsers = fmt.Sprintf("%s/%s", StockConfigPath, "parsers.conf")

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,