add support for throttle filter plugin

Signed-off-by: Nicholas Tate <[email protected]>

README.md

@@ -242,6 +242,7 @@ The list below shows supported plugins which are based on Fluent Bit v1.7.x+. Fo
     - [grep](docs/plugins/filter/grep.md)
     - [record modifier](docs/plugins/filter/recordmodifier.md)
     - [lua](docs/plugins/filter/lua.md)
+    - [throttle](docs/plugins/filter/throttle.md)
 - [Output](docs/crd.md#output)
     - [elasticsearch](docs/plugins/output/elasticsearch.md)
     - [file](docs/plugins/output/file.md)

api/fluentbitoperator/v1alpha2/filter_types.go

@@ -55,6 +55,8 @@ type FilterItem struct {
 	Parser *filter.Parser `json:"parser,omitempty"`
 	// Lua defines Lua Filter configuration.
 	Lua *filter.Lua `json:"lua,omitempty"`
+	// Throttle defines a Throttle configuration.
+	Throttle *filter.Throttle `json:"throttle,omitempty"`
 }
 
 // +kubebuilder:object:root=true

api/fluentbitoperator/v1alpha2/plugins/filter/throttle_types.go

@@ -0,0 +1,45 @@
+package filter
+
+import (
+	"fmt"
+
+	"kubesphere.io/fluentbit-operator/api/fluentbitoperator/v1alpha2/plugins"
+)
+
+// +kubebuilder:object:generate:=true
+
+// Throttle filter allows you to set the average rate of messages per internal, based on leaky bucket and sliding window algorithm.
+type Throttle struct {
+	// Rate is the amount of messages for the time.
+	Rate *int64 `json:"rate,omitempty"`
+	// Window is the amount of intervals to calculate average over.
+	Window *int64 `json:"window,omitempty"`
+	// Interval is the time interval expressed in "sleep" format. e.g. 3s, 1.5m, 0.5h, etc.
+	// +kubebuilder:validation:Pattern:="^\\d+(\\.[0-9]{0,2})?(s|m|h|d)?$"
+	Interval string `json:"interval,omitempty"`
+	// PrintStatus represents whether to print status messages with current rate and the limits to information logs.
+	PrintStatus *bool `json:"printStatus,omitempty"`
+}
+
+// Name is the name of the filter plugin.
+func (*Throttle) Name() string {
+	return "throttle"
+}
+
+// Params represents the config options for the filter plugin.
+func (k *Throttle) Params(_ plugins.SecretLoader) (*plugins.KVs, error) {
+	kvs := plugins.NewKVs()
+	if k.Rate != nil {
+		kvs.Insert("Rate", fmt.Sprint(*k.Rate))
+	}
+	if k.Window != nil {
+		kvs.Insert("Window", fmt.Sprint(*k.Window))
+	}
+	if k.Interval != "" {
+		kvs.Insert("Interval", k.Interval)
+	}
+	if k.PrintStatus != nil {
+		kvs.Insert("Print_Status", fmt.Sprint(*k.PrintStatus))
+	}
+	return kvs, nil
+}

chart/fluentbit-operator/crds/logging.kubesphere.io_filter.yaml

@@ -1,8 +1,11 @@
+
+---
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations:
     controller-gen.kubebuilder.io/version: v0.4.1
+  creationTimestamp: null
   name: filters.logging.kubesphere.io
 spec:
   group: logging.kubesphere.io
@@ -395,6 +398,29 @@ spec:
                             type: string
                           type: array
                       type: object
+                    throttle:
+                      description: Throttle defines a Throttle configuration.
+                      properties:
+                        interval:
+                          description: Interval is the time interval expressed in
+                            "sleep" format. e.g. 3s, 1.5m, 0.5h, etc.
+                          pattern: ^\d+(\.[0-9]({0,2})?(s|m|h|d)?$
+                          type: string
+                        printStatus:
+                          description: PrintStatus represents whether to print status
+                            messages with current rate and the limits to information
+                            logs.
+                          type: boolean
+                        rate:
+                          description: Rate is the amount of messages for the time.
+                          format: int64
+                          type: integer
+                        window:
+                          description: Window is the amount of intervals to calculate
+                            average over.
+                          format: int64
+                          type: integer
+                      type: object
                   type: object
                 type: array
               match:
@@ -414,4 +440,4 @@ status:
     kind: ""
     plural: ""
   conditions: []
-  storedVersions: []
+  storedVersions: []

config/crd/bases/logging.kubesphere.io_filters.yaml

@@ -398,6 +398,29 @@ spec:
                             type: string
                           type: array
                       type: object
+                    throttle:
+                      description: Throttle defines a Throttle configuration.
+                      properties:
+                        interval:
+                          description: Interval is the time interval expressed in
+                            "sleep" format. e.g. 3s, 1.5m, 0.5h, etc.
+                          pattern: ^\d+(\.[0-9]{0,2})?(s|m|h|d)?$
+                          type: string
+                        printStatus:
+                          description: PrintStatus represents whether to print status
+                            messages with current rate and the limits to information
+                            logs.
+                          type: boolean
+                        rate:
+                          description: Rate is the amount of messages for the time.
+                          format: int64
+                          type: integer
+                        window:
+                          description: Window is the amount of intervals to calculate
+                            average over.
+                          format: int64
+                          type: integer
+                      type: object
                   type: object
                 type: array
               match:

docs/plugins/filter/throttle.md

@@ -0,0 +1,11 @@
+# Throttle
+
+The Throttle Filter plugin sets the average Rate of messages per Interval, based on leaky bucket and sliding window algorithm. In case of overflood, it will leak within certain rate.
+
+
+| Field | Description | Scheme |
+| ----- | ----------- | ------ |
+| rate | Amount of messages for the time. | *int64 |
+| window | Amount of intervals to calculate average over. | *int64 |
+| interval | Time interval, expressed in "sleep" format. e.g 3s, 1.5m, 0.5h etc | string |
+| printStatus | Whether to print status messages with current rate and the limits to information logs | *bool |

manifests/setup/fluentbit-operator-crd.yaml

@@ -395,6 +395,29 @@ spec:
                             type: string
                           type: array
                       type: object
+                    throttle:
+                      description: Throttle defines a Throttle configuration.
+                      properties:
+                        interval:
+                          description: Interval is the time interval expressed in
+                            "sleep" format. e.g. 3s, 1.5m, 0.5h, etc.
+                          pattern: ^\d+(\.[0-9]{0,2})?(s|m|h|d)?$
+                          type: string
+                        printStatus:
+                          description: PrintStatus represents whether to print status
+                            messages with current rate and the limits to information
+                            logs.
+                          type: boolean
+                        rate:
+                          description: Rate is the amount of messages for the time.
+                          format: int64
+                          type: integer
+                        window:
+                          description: Window is the amount of intervals to calculate
+                            average over.
+                          format: int64
+                          type: integer
+                      type: object
                   type: object
                 type: array
               match: