Configure default_route option from fluent-plugin-label-router (#424)

* Define a DefaultFlow on the Logging object

Co-authored-by: Sándor Guba <[email protected]>

Author: PhilGrayson

charts/logging-operator/crds/logging.banzaicloud.io_loggings.yaml

@@ -101,7 +101,18 @@ func (l *LoggingResources) CreateModel() (*types.Builder, error) {
 			return nil, err
 		}
 	}
-	if len(l.Flows) == 0 && len(l.ClusterFlows) == 0 {
+	if l.logging.Spec.DefaultFlowSpec != nil {
+		flow, err := l.CreateFlowFromCustomResource(l.logging.Spec.DefaultFlowSpec)
+		if err != nil {
+			// TODO set flow status to error?
+			return nil, err
+		}
+		err = system.RegisterDefaultFlow(flow)
+		if err != nil {
+			return nil, err
+		}
+	}
+	if len(l.Flows) == 0 && len(l.ClusterFlows) == 0 && l.logging.Spec.DefaultFlowSpec == nil {
 		l.logger.Info("no flows found, generating empty model")
 	}
 	return system, nil
@@ -164,10 +175,24 @@ func GetFlowMatchFromSpec(namespace string, matches interface{}) ([]types.FlowMa
 	return flowMatches, nil
 }
 
-func FlowDispatcher(flowCr interface{}) (*CommonFlow, error) {
+func (l *LoggingResources) FlowDispatcher(flowCr interface{}) (*CommonFlow, error) {
 	var commonFlow *CommonFlow
 	var err error
 	switch f := flowCr.(type) {
+	case *v1beta1.DefaultFlowSpec:
+		id := fmt.Sprintf("logging:%s:%s", l.logging.Namespace, l.logging.Name)
+		flow, err := types.NewFlow([]types.FlowMatch{}, id, l.logging.Name, l.logging.Namespace)
+		if err != nil {
+			return nil, err
+		}
+		return &CommonFlow{
+			Name:       l.logging.Name,
+			Namespace:  l.logging.Namespace,
+			Scope:      "",
+			OutputRefs: f.OutputRefs,
+			Filters:    f.Filters,
+			Flow:       flow,
+		}, nil
 	case v1beta1.ClusterFlow:
 		var matches []types.FlowMatch
 		commonFlow = &CommonFlow{
@@ -241,7 +266,7 @@ func FlowDispatcher(flowCr interface{}) (*CommonFlow, error) {
 }
 
 func (l *LoggingResources) CreateFlowFromCustomResource(flowCr interface{}) (*types.Flow, error) {
-	commonFlow, err := FlowDispatcher(flowCr)
+	commonFlow, err := l.FlowDispatcher(flowCr)
 	if err != nil {
 		return nil, err
 	}

config/crd/bases/logging.banzaicloud.io_loggings.yaml

@@ -41,13 +41,14 @@ type _metaLoggingSpec interface{}
 
 // LoggingSpec defines the desired state of Logging
 type LoggingSpec struct {
-	LoggingRef              string         `json:"loggingRef,omitempty"`
-	FlowConfigCheckDisabled bool           `json:"flowConfigCheckDisabled,omitempty"`
-	FlowConfigOverride      string         `json:"flowConfigOverride,omitempty"`
-	FluentbitSpec           *FluentbitSpec `json:"fluentbit,omitempty"`
-	FluentdSpec             *FluentdSpec   `json:"fluentd,omitempty"`
-	WatchNamespaces         []string       `json:"watchNamespaces,omitempty"`
-	ControlNamespace        string         `json:"controlNamespace"`
+	LoggingRef              string           `json:"loggingRef,omitempty"`
+	FlowConfigCheckDisabled bool             `json:"flowConfigCheckDisabled,omitempty"`
+	FlowConfigOverride      string           `json:"flowConfigOverride,omitempty"`
+	FluentbitSpec           *FluentbitSpec   `json:"fluentbit,omitempty"`
+	FluentdSpec             *FluentdSpec     `json:"fluentd,omitempty"`
+	DefaultFlowSpec         *DefaultFlowSpec `json:"defaultFlow,omitempty"`
+	WatchNamespaces         []string         `json:"watchNamespaces,omitempty"`
+	ControlNamespace        string           `json:"controlNamespace"`
 
 	// EnableRecreateWorkloadOnImmutableFieldChange enables the operator to recreate the
 	// fluentbit daemonset and the fluentd statefulset (and possibly other resource in the future)
@@ -83,6 +84,14 @@ type LoggingList struct {
 	Items           []Logging `json:"items"`
 }
 
+// +kubebuilder:object:generate=true
+
+// DefaultFlowSpec is a Flow for logs that did not match any other Flow
+type DefaultFlowSpec struct {
+	Filters    []Filter `json:"filters,omitempty"`
+	OutputRefs []string `json:"outputRefs"`
+}
+
 // SetDefaults fill empty attributes
 func (l *Logging) SetDefaults() (*Logging, error) {
 	copy := l.DeepCopy()

pkg/resources/model/system.go

@@ -42,6 +42,17 @@ func (s *Builder) RegisterFlow(f *Flow) error {
 	return nil
 }
 
+func (s *Builder) RegisterDefaultFlow(f *Flow) error {
+	for _, e := range s.flows {
+		if e.FlowLabel == f.FlowLabel {
+			return errors.New("Flow already exists")
+		}
+	}
+	s.flows = append(s.flows, f)
+	s.router.Params["default_route"] = f.FlowLabel
+	return nil
+}
+
 func (s *Builder) Build() (*System, error) {
 	return &System{
 		Input:  s.input,