Adding Fluentbit's unified networking interface (#1217)

* Adding unified networking interface for fluentbit configuration Signed-off-by: localleon <me@lrau.xyz> * Renaming networking plugin Signed-off-by: localleon <me@lrau.xyz> * Adding suppport for networking interface to all fluentbit plugins that support the networking plugin (e.g where we currently already have the TLS interface) Signed-off-by: localleon <me@lrau.xyz> * Renamed *plugin.networking json-key for better readability in CRDs Signed-off-by: localleon <me@lrau.xyz> --------- Signed-off-by: localleon <me@lrau.xyz>
fluent · Jun 25, 2024 · db04760 · db04760
1 parent f46a1a5
commit db04760
Show file tree

Hide file tree

Showing 36 changed files with 6,303 additions and 0 deletions.
diff --git a/apis/fluentbit/v1alpha2/plugins/net_types.go b/apis/fluentbit/v1alpha2/plugins/net_types.go
@@ -0,0 +1,71 @@
+package plugins
+
+import (
+	"fmt"
+
+	"github.com/fluent/fluent-operator/v2/apis/fluentbit/v1alpha2/plugins/params"
+)
+
+// +kubebuilder:object:generate:=true
+
+// Fluent Bit implements a unified networking interface that is exposed to components like plugins. These are the functions from https://docs.fluentbit.io/manual/administration/networking and can be used on various output plugins
+type Networking struct {
+	// Set maximum time expressed in seconds to wait for a TCP connection to be established, this include the TLS handshake time.
+	ConnectTimeout *int32 `json:"connectTimeout,omitempty"`
+	// On connection timeout, specify if it should log an error. When disabled, the timeout is logged as a debug message.
+	ConnectTimeoutLogError *bool `json:"connectTimeoutLogError,omitempty"`
+	// Select the primary DNS connection type (TCP or UDP).
+	// +kubebuilder:validation:Enum:="TCP";"UDP"
+	DNSMode *string `json:"DNSMode,omitempty"`
+	// Prioritize IPv4 DNS results when trying to establish a connection.
+	DNSPreferIPv4 *bool `json:"DNSPreferIPv4,omitempty"`
+	// Select the primary DNS resolver type (LEGACY or ASYNC).
+	// +kubebuilder:validation:Enum:="LEGACY";"ASYNC"
+	DNSResolver *string `json:"DNSResolver,omitempty"`
+	// Enable or disable connection keepalive support. Accepts a boolean value: on / off.
+	// +kubebuilder:validation:Enum:="on";"off"
+	Keepalive *string `json:"keepalive,omitempty"`
+	// Set maximum time expressed in seconds for an idle keepalive connection.
+	KeepaliveIdleTimeout *int32 `json:"keepaliveIdleTimeout,omitempty"`
+	// Set maximum number of times a keepalive connection can be used before it is retired.
+	KeepaliveMaxRecycle *int32 `json:"keepaliveMaxRecycle,omitempty"`
+	// Set maximum number of TCP connections that can be established per worker.
+	MaxWorkerConnections *int32 `json:"maxWorkerConnections,omitempty"`
+	// Specify network address to bind for data traffic.
+	SourceAddress *string `json:"sourceAddress,omitempty"`
+}
+
+func (t *Networking) Params(sl SecretLoader) (*params.KVs, error) {
+	kvs := params.NewKVs()
+	if t.ConnectTimeout != nil {
+		kvs.Insert("net.connect_timeout", fmt.Sprint(*t.ConnectTimeout))
+	}
+	if t.ConnectTimeoutLogError != nil {
+		kvs.Insert("net.connect_timeout_log_error", fmt.Sprint(*t.ConnectTimeoutLogError))
+	}
+	if t.DNSMode != nil {
+		kvs.Insert("net.dns.mode", *t.DNSMode)
+	}
+	if t.DNSPreferIPv4 != nil {
+		kvs.Insert("net.dns.prefer_ipv4", fmt.Sprint(*t.DNSPreferIPv4))
+	}
+	if t.DNSResolver != nil {
+		kvs.Insert("net.dns.prefer_ipv4", *t.DNSResolver)
+	}
+	if t.Keepalive != nil {
+		kvs.Insert("net.keepalive", *t.Keepalive)
+	}
+	if t.KeepaliveIdleTimeout != nil {
+		kvs.Insert("net.keepalive_idle_timeout", fmt.Sprint(*t.KeepaliveIdleTimeout))
+	}
+	if t.KeepaliveMaxRecycle != nil {
+		kvs.Insert("net.keepalive_max_recycle", fmt.Sprint(*t.KeepaliveMaxRecycle))
+	}
+	if t.MaxWorkerConnections != nil {
+		kvs.Insert("net.max_worker_connections", fmt.Sprint(*t.MaxWorkerConnections))
+	}
+	if t.SourceAddress != nil {
+		kvs.Insert("net.source_address", *t.SourceAddress)
+	}
+	return kvs, nil
+}
diff --git a/apis/fluentbit/v1alpha2/plugins/output/azure_blob_types.go b/apis/fluentbit/v1alpha2/plugins/output/azure_blob_types.go
@@ -31,6 +31,8 @@ type AzureBlob struct {
 	Endpoint string `json:"endpoint,omitempty"`
 	// Enable/Disable TLS Encryption. Azure services require TLS to be enabled.
 	*plugins.TLS `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 }
 
 // Name implement Section() method

diff --git a/apis/fluentbit/v1alpha2/plugins/output/elasticsearch_types.go b/apis/fluentbit/v1alpha2/plugins/output/elasticsearch_types.go
@@ -100,6 +100,8 @@ type Elasticsearch struct {
 	// When enabled, mapping types is removed and Type option is ignored. Types are deprecated in APIs in v7.0. This options is for v7.0 or later.
 	SuppressTypeName string `json:"suppressTypeName,omitempty"`
 	*plugins.TLS     `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 	// Limit the maximum number of Chunks in the filesystem for the current output logical destination.
 	TotalLimitSize string `json:"totalLimitSize,omitempty"`
 }

diff --git a/apis/fluentbit/v1alpha2/plugins/output/forward_types.go b/apis/fluentbit/v1alpha2/plugins/output/forward_types.go
@@ -41,6 +41,8 @@ type Forward struct {
 	// Default value of the auto-generated certificate common name (CN).
 	SelfHostname string `json:"selfHostname,omitempty"`
 	*plugins.TLS `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 }
 
 func (_ *Forward) Name() string {

diff --git a/apis/fluentbit/v1alpha2/plugins/output/gelf_types.go b/apis/fluentbit/v1alpha2/plugins/output/gelf_types.go
@@ -36,6 +36,8 @@ type Gelf struct {
 	// If transport protocol is udp, it defines if UDP packets should be compressed.
 	Compress     *bool `json:"compress,omitempty"`
 	*plugins.TLS `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 }
 
 func (_ *Gelf) Name() string {

diff --git a/apis/fluentbit/v1alpha2/plugins/output/http_types.go b/apis/fluentbit/v1alpha2/plugins/output/http_types.go
@@ -61,6 +61,8 @@ type HTTP struct {
 	// HTTP output plugin supports TTL/SSL, for more details about the properties available
 	// and general configuration, please refer to the TLS/SSL section.
 	*plugins.TLS `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 }
 
 // implement Name method

diff --git a/apis/fluentbit/v1alpha2/plugins/output/influxdb_types.go b/apis/fluentbit/v1alpha2/plugins/output/influxdb_types.go
@@ -45,6 +45,8 @@ type InfluxDB struct {
 	// Key of the string array optionally contained within each log record that contains tag keys for that record
 	TagsListKey  string `json:"tagListKey,omitempty"`
 	*plugins.TLS `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 }
 
 // Name implement Section() method

diff --git a/apis/fluentbit/v1alpha2/plugins/output/loki_types.go b/apis/fluentbit/v1alpha2/plugins/output/loki_types.go
@@ -52,6 +52,8 @@ type Loki struct {
 	// The value of the key is set as X-Scope-OrgID of HTTP header. It is useful to set Tenant ID dynamically.
 	TenantIDKey  string `json:"tenantIDKey,omitempty"`
 	*plugins.TLS `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 }
 
 // implement Section() method

diff --git a/apis/fluentbit/v1alpha2/plugins/output/open_search_types.go b/apis/fluentbit/v1alpha2/plugins/output/open_search_types.go
@@ -2,6 +2,7 @@ package output
 
 import (
 	"fmt"
+
 	"github.com/fluent/fluent-operator/v2/apis/fluentbit/v1alpha2/plugins"
 	"github.com/fluent/fluent-operator/v2/apis/fluentbit/v1alpha2/plugins/params"
 )
@@ -94,6 +95,8 @@ type OpenSearch struct {
 	// Enables dedicated thread(s) for this output. Default value is set since version 1.8.13. For previous versions is 0.
 	Workers      *int32 `json:"Workers,omitempty"`
 	*plugins.TLS `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 	// Limit the maximum number of Chunks in the filesystem for the current output logical destination.
 	TotalLimitSize string `json:"totalLimitSize,omitempty"`
 }

diff --git a/apis/fluentbit/v1alpha2/plugins/output/open_telemetry_types.go b/apis/fluentbit/v1alpha2/plugins/output/open_telemetry_types.go
@@ -38,6 +38,8 @@ type OpenTelemetry struct {
 	// This allows you to add custom labels to all metrics exposed through the OpenTelemetry exporter. You may have multiple of these fields.
 	AddLabel     map[string]string `json:"addLabel,omitempty"`
 	*plugins.TLS `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 }
 
 // Name implement Section() method

diff --git a/apis/fluentbit/v1alpha2/plugins/output/prometheus_remote_write_types.go b/apis/fluentbit/v1alpha2/plugins/output/prometheus_remote_write_types.go
@@ -39,6 +39,8 @@ type PrometheusRemoteWrite struct {
 	Workers *int32 `json:"workers,omitempty"`
 
 	*plugins.TLS `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 }
 
 // implement Section() method

diff --git a/apis/fluentbit/v1alpha2/plugins/output/splunk_types.go b/apis/fluentbit/v1alpha2/plugins/output/splunk_types.go
@@ -61,6 +61,8 @@ type Splunk struct {
 	// Enables dedicated thread(s) for this output. Default value `2` is set since version 1.8.13. For previous versions is 0.
 	Workers      *int32 `json:"Workers,omitempty"`
 	*plugins.TLS `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 }
 
 // Name implement Section() method

diff --git a/apis/fluentbit/v1alpha2/plugins/output/syslog_types.go b/apis/fluentbit/v1alpha2/plugins/output/syslog_types.go
@@ -43,6 +43,8 @@ type Syslog struct {
 	// Syslog output plugin supports TTL/SSL, for more details about the properties available
 	// and general configuration, please refer to the TLS/SSL section.
 	*plugins.TLS `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 }
 
 func (_ *Syslog) Name() string {

diff --git a/apis/fluentbit/v1alpha2/plugins/output/tcp_types.go b/apis/fluentbit/v1alpha2/plugins/output/tcp_types.go
@@ -30,6 +30,8 @@ type TCP struct {
 	// +kubebuilder:validation:Enum:=double;epoch;iso8601
 	JsonDateFormat string `json:"jsonDateFormat,omitempty"`
 	*plugins.TLS   `json:"tls,omitempty"`
+	// Include fluentbit networking options for this output-plugin
+	*plugins.Networking `json:"networking,omitempty"`
 }
 
 func (_ *TCP) Name() string {
@@ -60,5 +62,12 @@ func (t *TCP) Params(sl plugins.SecretLoader) (*params.KVs, error) {
 		}
 		kvs.Merge(tls)
 	}
+	if t.Networking != nil {
+		net, err := t.Networking.Params(sl)
+		if err != nil {
+			return nil, err
+		}
+		kvs.Merge(net)
+	}
 	return kvs, nil
 }
diff --git a/apis/fluentbit/v1alpha2/plugins/output/zz_generated.deepcopy.go b/apis/fluentbit/v1alpha2/plugins/output/zz_generated.deepcopy.go