add systemd input and lua filter support

Signed-off-by: wanjunlei <[email protected]>

Author: wanjunlei

api/v1alpha2/filter_types.go

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

api/v1alpha2/fluentbitconfig_types.go

@@ -167,3 +167,21 @@ func (cfg FluentBitConfig) RenderParserConfig(sl plugins.SecretLoader, parsers P
 
 	return buf.String(), nil
 }
+
+func (cfg FluentBitConfig) RenderLuaScript(cl plugins.ConfigMapLoader, filters FilterList) (map[string]string, error) {
+
+	scripts := make(map[string]string)
+	for _, f := range filters.Items {
+		for _, p := range f.Spec.FilterItems {
+			if p.Lua != nil {
+				script, err := cl.LoadConfigMap(p.Lua.Script)
+				if err != nil {
+					return nil, err
+				}
+				scripts[p.Lua.Script.Key] = script
+			}
+		}
+	}
+
+	return scripts, nil
+}

api/v1alpha2/input_types.go

@@ -33,6 +33,8 @@ type InputSpec struct {
 	Dummy *input.Dummy `json:"dummy,omitempty"`
 	// Tail defines Tail Input configuration.
 	Tail *input.Tail `json:"tail,omitempty"`
+	// Systemd defines Systemd Input configuration.
+	Systemd *input.Systemd `json:"systemd,omitempty"`
 }
 
 // +kubebuilder:object:root=true

api/v1alpha2/plugins/configmap_types.go

@@ -0,0 +1,35 @@
+package plugins
+
+import (
+	"context"
+	"fmt"
+	"github.com/go-openapi/errors"
+	"k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"strings"
+)
+
+type ConfigMapLoader struct {
+	client    client.Client
+	namespace string
+}
+
+func NewConfigMapLoader(c client.Client, ns string) ConfigMapLoader {
+	return ConfigMapLoader{
+		client:    c,
+		namespace: ns,
+	}
+}
+
+func (cl ConfigMapLoader) LoadConfigMap(selector v1.ConfigMapKeySelector) (string, error) {
+	var configMap v1.ConfigMap
+	if err := cl.client.Get(context.Background(), client.ObjectKey{Name: selector.Name, Namespace: cl.namespace}, &configMap); err != nil {
+		return "", err
+	}
+
+	if v, ok := configMap.Data[selector.Key]; !ok {
+		return "", errors.NotFound(fmt.Sprintf("The key %s is not found.", selector.Key))
+	} else {
+		return strings.TrimSuffix(fmt.Sprintf("%s", v), "\n"), nil
+	}
+}

api/v1alpha2/plugins/filter/lua_types.go

@@ -0,0 +1,57 @@
+package filter
+
+import (
+	v1 "k8s.io/api/core/v1"
+	"kubesphere.io/fluentbit-operator/api/v1alpha2/plugins"
+	"kubesphere.io/fluentbit-operator/pkg/utils"
+	"strconv"
+)
+
+// +kubebuilder:object:generate:=true
+
+// The Lua Filter allows you to modify the incoming records using custom Lua Scripts.
+type Lua struct {
+
+	// Path to the Lua script that will be used.
+	Script v1.ConfigMapKeySelector `json:"script"`
+	// Lua function name that will be triggered to do filtering.
+	// It's assumed that the function is declared inside the Script defined above.
+	Call string `json:"call"`
+	// If these keys are matched, the fields are converted to integer.
+	// If more than one key, delimit by space.
+	// Note that starting from Fluent Bit v1.6 integer data types are preserved
+	// and not converted to double as in previous versions.
+	TypeIntKey []string `json:"typeIntKey,omitempty"`
+	// If enabled, Lua script will be executed in protected mode.
+	// It prevents to crash when invalid Lua script is executed. Default is true.
+	ProtectedMode *bool `json:"protectedMode,omitempty"`
+	// By default when the Lua script is invoked, the record timestamp is passed as a
+	// Floating number which might lead to loss precision when the data is converted back.
+	// If you desire timestamp precision enabling this option will pass the timestamp as
+	// a Lua table with keys sec for seconds since epoch and nsec for nanoseconds.
+	TimeAsTable bool `json:"timeAsTable,omitempty"`
+}
+
+func (l *Lua) Name() string {
+	return "lua"
+}
+
+func (l *Lua) Params(_ plugins.SecretLoader) (*plugins.KVs, error) {
+	kvs := plugins.NewKVs()
+
+	kvs.Insert("script", "/fluent-bit/config/"+l.Script.Key)
+	kvs.Insert("call", l.Call)
+
+	if l.TypeIntKey != nil && len(l.TypeIntKey) > 0 {
+		kvs.Insert("type_int_key", utils.ConcatString(l.TypeIntKey, " "))
+	}
+
+	if l.ProtectedMode != nil {
+		kvs.Insert("protected_mode", strconv.FormatBool(*l.ProtectedMode))
+	}
+
+	if l.TimeAsTable {
+		kvs.Insert("time_as_table", "true")
+	}
+	return kvs, nil
+}

api/v1alpha2/plugins/filter/zz_generated.deepcopy.go

@@ -0,0 +1,88 @@
+package input
+
+import (
+	"kubesphere.io/fluentbit-operator/api/v1alpha2/plugins"
+)
+
+// +kubebuilder:object:generate:=true
+
+// The Systemd input plugin allows to collect log messages from the Journald daemon on Linux environments.
+type Systemd struct {
+	// Optional path to the Systemd journal directory,
+	// if not set, the plugin will use default paths to read local-only logs.
+	Path string `json:"path,omitempty"`
+	// Specify the database file to keep track of monitored files and offsets.
+	DB string `json:"db,omitempty"`
+	// Set a default synchronization (I/O) method. values: Extra, Full, Normal, Off.
+	// This flag affects how the internal SQLite engine do synchronization to disk,
+	// for more details about each option please refer to this section.
+	// note: this option was introduced on Fluent Bit v1.4.6.
+	// +kubebuilder:validation:Enum:=Extra;Full;Normal;Off
+	DBSync string `json:"dbSync,omitempty"`
+	// The tag is used to route messages but on Systemd plugin there is an extra functionality:
+	// if the tag includes a star/wildcard, it will be expanded with the Systemd Unit file (e.g: host.* => host.UNIT_NAME).
+	Tag string `json:"tag,omitempty"`
+	// Set a maximum number of fields (keys) allowed per record.
+	MaxFields int `json:"maxFields,omitempty"`
+	// When Fluent Bit starts, the Journal might have a high number of logs in the queue.
+	// In order to avoid delays and reduce memory usage, this option allows to specify the maximum number of log entries that can be processed per round.
+	// Once the limit is reached, Fluent Bit will continue processing the remaining log entries once Journald performs the notification.
+	MaxEntries int `json:"maxEntries,omitempty"`
+	// Allows to perform a query over logs that contains a specific Journald key/value pairs, e.g: _SYSTEMD_UNIT=UNIT.
+	// The Systemd_Filter option can be specified multiple times in the input section to apply multiple filters as required.
+	SystemdFilter []string `json:"systemdFilter,omitempty"`
+	// Define the filter type when Systemd_Filter is specified multiple times. Allowed values are And and Or.
+	// With And a record is matched only when all of the Systemd_Filter have a match.
+	// With Or a record is matched when any of the Systemd_Filter has a match.
+	// +kubebuilder:validation:Enum:=And;Or
+	SystemdFilterType string `json:"systemdFilterType,omitempty"`
+	// Start reading new entries. Skip entries already stored in Journald.
+	// +kubebuilder:validation:Enum:=on;off
+	ReadFromTail string `json:"readFromTail,omitempty"`
+	// Remove the leading underscore of the Journald field (key). For example the Journald field _PID becomes the key PID.
+	// +kubebuilder:validation:Enum:=on;off
+	StripUnderscores string `json:"stripUnderscores,omitempty"`
+}
+
+func (_ *Systemd) Name() string {
+	return "systemd"
+}
+
+func (s *Systemd) Params(_ plugins.SecretLoader) (*plugins.KVs, error) {
+	kvs := plugins.NewKVs()
+
+	if s.Path != "" {
+		kvs.Insert("Path", s.Path)
+	}
+	if s.DB != "" {
+		kvs.Insert("DB", s.DB)
+	}
+	if s.DBSync != "" {
+		kvs.Insert("DB.Sync", s.DBSync)
+	}
+	if s.Tag != "" {
+		kvs.Insert("Tag", s.Tag)
+	}
+	if s.MaxFields > 0 {
+		kvs.Insert("Max_Fields", string(rune(s.MaxFields)))
+	}
+	if s.MaxEntries > 0 {
+		kvs.Insert("Max_Entries", string(rune(s.MaxEntries)))
+	}
+	if s.SystemdFilter != nil && len(s.SystemdFilter) > 0 {
+		for _, v := range s.SystemdFilter {
+			kvs.Insert("Systemd_Filter", v)
+		}
+	}
+	if s.SystemdFilterType != "" {
+		kvs.Insert("Systemd_Filter_Type", s.SystemdFilterType)
+	}
+	if s.ReadFromTail != "" {
+		kvs.Insert("Read_From_Tail", s.ReadFromTail)
+	}
+	if s.StripUnderscores != "" {
+		kvs.Insert("Strip_Underscores", s.StripUnderscores)
+	}
+
+	return kvs, nil
+}

api/v1alpha2/plugins/input/systemd_types.go

@@ -147,6 +147,56 @@ spec:
                           Journald format.
                         type: boolean
                     type: object
+                  lua:
+                    description: Lua defines Lua Filter configuration.
+                    properties:
+                      call:
+                        description: Lua function name that will be triggered to do
+                          filtering. It's assumed that the function is declared inside
+                          the Script defined above.
+                        type: string
+                      protectedMode:
+                        description: If enabled, Lua script will be executed in protected
+                          mode. It prevents to crash when invalid Lua script is executed.
+                          Default is true.
+                        type: boolean
+                      script:
+                        description: Path to the Lua script that will be used.
+                        properties:
+                          key:
+                            description: The key to select.
+                            type: string
+                          name:
+                            description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                              TODO: Add other useful fields. apiVersion, kind, uid?'
+                            type: string
+                          optional:
+                            description: Specify whether the ConfigMap or its key
+                              must be defined
+                            type: boolean
+                        required:
+                        - key
+                        type: object
+                      timeAsTable:
+                        description: By default when the Lua script is invoked, the
+                          record timestamp is passed as a Floating number which might
+                          lead to loss precision when the data is converted back.
+                          If you desire timestamp precision enabling this option will
+                          pass the timestamp as a Lua table with keys sec for seconds
+                          since epoch and nsec for nanoseconds.
+                        type: boolean
+                      typeIntKey:
+                        description: If these keys are matched, the fields are converted
+                          to integer. If more than one key, delimit by space. Note
+                          that starting from Fluent Bit v1.6 integer data types are
+                          preserved and not converted to double as in previous versions.
+                        items:
+                          type: string
+                        type: array
+                    required:
+                    - call
+                    - script
+                    type: object
                   modify:
                     description: Modify defines Modify Filter configuration.
                     properties: