fluentbit: enable multiple FluentbitAgents for a single logging resource; use the FluentbitAgent resource's name as the default hostPath for volumes; fix the for loop variable with a deepcopy

Signed-off-by: Peter Wilcsinszky <[email protected]>

Author: pepov

controllers/logging/logging_controller.go

@@ -186,35 +186,38 @@ func (r *LoggingReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	}
 
 	switch len(loggingResources.Fluentbits) {
-	case 1:
-		if logging.Spec.FluentbitSpec != nil {
-			return ctrl.Result{}, errors.New("fluentbit has to be removed from the logging resource before the new FluentbitAgent can be reconciled")
-		}
-		reconcilers = append(reconcilers, fluentbit.New(
-			r.Client,
-			r.Log,
-			&logging,
-			reconcilerOpts,
-			&loggingResources.Fluentbits[0].Spec,
-			loggingDataProvider,
-			loggingv1beta1.NewStandaloneFluentbitNameProvider(&loggingResources.Fluentbits[0]),
-		).Reconcile)
 	case 0:
 		// check for legacy definition
 		log.Info("WARNING fluentbit definition inside the Logging resource is deprecated and will be removed in the next major release")
 		if logging.Spec.FluentbitSpec != nil {
+			nameProvider := loggingv1beta1.NewLegacyFluentbitNameProvider(&logging)
 			reconcilers = append(reconcilers, fluentbit.New(
 				r.Client,
-				r.Log,
+				r.Log.WithName(nameProvider.ComponentName("logging-legacy")),
 				&logging,
 				reconcilerOpts,
 				logging.Spec.FluentbitSpec,
 				loggingDataProvider,
-				loggingv1beta1.NewLegacyFluentbitNameProvider(&logging),
+				nameProvider,
 			).Reconcile)
 		}
 	default:
-		return ctrl.Result{}, errors.New("cannot handle more than one FluentbitAgent for the same Logging resource")
+		if logging.Spec.FluentbitSpec != nil {
+			return ctrl.Result{}, errors.New("fluentbit has to be removed from the logging resource before the new FluentbitAgent can be reconciled")
+		}
+		for _, f := range loggingResources.Fluentbits {
+			f := f.DeepCopy()
+			nameProvider := loggingv1beta1.NewStandaloneFluentbitNameProvider(f)
+			reconcilers = append(reconcilers, fluentbit.New(
+				r.Client,
+				r.Log.WithName(nameProvider.ComponentName("fluentbit-agent")),
+				&logging,
+				reconcilerOpts,
+				&f.Spec,
+				loggingDataProvider,
+				nameProvider,
+			).Reconcile)
+		}
 	}
 
 	if len(logging.Spec.NodeAgents) > 0 || len(loggingResources.NodeAgents) > 0 {

pkg/resources/fluentbit/configsecret.go

@@ -246,8 +246,7 @@ func (r *Reconciler) configSecret() (runtime.Object, reconciler.DesiredState, er
 		// If MultilineParser is set, remove other parser fields
 		// See https://docs.fluentbit.io/manual/pipeline/inputs/tail#multiline-core-v1.8
 
-		log.Log.Info("Notice: MultilineParser is enabled. Disabling other parser options",
-			"logging", r.Logging.Name)
+		r.logger.Info("Notice: MultilineParser is enabled. Disabling other parser options")
 
 		inputTail.Parser = ""
 		inputTail.ParserFirstline = ""
@@ -305,25 +304,25 @@ func (r *Reconciler) configSecret() (runtime.Object, reconciler.DesiredState, er
 			input.FluentForwardOutput.Network = newFluentbitNetwork(*r.fluentbitSpec.Network)
 		}
 
-		fluentdReplicas, err := r.loggingDataProvider.GetReplicaCount(context.TODO())
+		aggregatorReplicas, err := r.loggingDataProvider.GetReplicaCount(context.TODO())
 		if err != nil {
 			return nil, nil, errors.WrapIf(err, "getting replica count for fluentd")
 		}
 
-		if r.fluentbitSpec.Network == nil && utils.PointerToInt32(fluentdReplicas) > 1 {
+		if r.fluentbitSpec.Network == nil && utils.PointerToInt32(aggregatorReplicas) > 1 {
 			input.FluentForwardOutput.Network.KeepaliveSet = true
 			input.FluentForwardOutput.Network.Keepalive = true
 			input.FluentForwardOutput.Network.KeepaliveIdleTimeoutSet = true
 			input.FluentForwardOutput.Network.KeepaliveIdleTimeout = 30
 			input.FluentForwardOutput.Network.KeepaliveMaxRecycleSet = true
 			input.FluentForwardOutput.Network.KeepaliveMaxRecycle = 100
-			log.Log.Info("Notice: Because the Fluentd statefulset has been scaled, we've made some changes in the fluentbit network config too. We advice to revise these default configurations.")
+			log.Log.Info("Notice: fluentbit `network` settings have been configured automatically to adapt to multiple aggregator replicas. Configure it manually to avoid this notice.")
 		}
 
 		if r.fluentbitSpec.EnableUpstream {
 			input.FluentForwardOutput.Upstream.Enabled = true
 			input.FluentForwardOutput.Upstream.Config.Name = "fluentd-upstream"
-			for i := int32(0); i < utils.PointerToInt32(fluentdReplicas); i++ {
+			for i := int32(0); i < utils.PointerToInt32(aggregatorReplicas); i++ {
 				input.FluentForwardOutput.Upstream.Config.Nodes = append(input.FluentForwardOutput.Upstream.Config.Nodes, r.generateUpstreamNode(i))
 			}
 		}

pkg/resources/fluentbit/daemonset.go

@@ -94,9 +94,9 @@ func (r *Reconciler) daemonSet() (runtime.Object, reconciler.DesiredState, error
 	}
 
 	r.fluentbitSpec.PositionDB.WithDefaultHostPath(
-		fmt.Sprintf(v1beta1.HostPath, r.Logging.Name, TailPositionVolume))
+		fmt.Sprintf(v1beta1.HostPath, r.nameProvider.Name(), TailPositionVolume))
 	r.fluentbitSpec.BufferStorageVolume.WithDefaultHostPath(
-		fmt.Sprintf(v1beta1.HostPath, r.Logging.Name, BufferStorageVolume))
+		fmt.Sprintf(v1beta1.HostPath, r.nameProvider.Name(), BufferStorageVolume))
 
 	if err := r.fluentbitSpec.PositionDB.ApplyVolumeForPodSpec(TailPositionVolume, containerName, "/tail-db", &desired.Spec.Template.Spec); err != nil {
 		return desired, reconciler.StatePresent, err

pkg/resources/fluentbit/fluentbit.go

@@ -52,8 +52,13 @@ func generateLoggingRefLabels(loggingRef string) map[string]string {
 }
 
 func (r *Reconciler) getFluentBitLabels() map[string]string {
-	return util.MergeLabels(r.fluentbitSpec.Labels, map[string]string{
-		"app.kubernetes.io/name": "fluentbit"}, generateLoggingRefLabels(r.Logging.GetName()))
+	return util.MergeLabels(
+		r.fluentbitSpec.Labels,
+		map[string]string{
+			"app.kubernetes.io/instance": r.nameProvider.Name(),
+			"app.kubernetes.io/name":     "fluentbit",
+		},
+		generateLoggingRefLabels(r.Logging.GetName()))
 }
 
 func (r *Reconciler) getServiceAccount() string {
@@ -64,7 +69,11 @@ func (r *Reconciler) getServiceAccount() string {
 }
 
 type NameProvider interface {
+	// ComponentName provides a qualified name using (Name + "-" + name)
 	ComponentName(name string) string
+	// Name returns the name of the resource, that is owning fluentbit
+	// It is Logging.Name for legacy but the resource's name for FluentbitAgent
+	Name() string
 	OwnerRef() v1.OwnerReference
 }
 
@@ -75,7 +84,8 @@ type DesiredObject struct {
 
 // Reconciler holds info what resource to reconcile
 type Reconciler struct {
-	*reconciler.GenericResourceReconciler
+	resourceReconciler  *reconciler.GenericResourceReconciler
+	logger              logr.Logger
 	Logging             *v1beta1.Logging
 	configs             map[string][]byte
 	fluentbitSpec       *v1beta1.FluentbitSpec
@@ -92,11 +102,12 @@ func New(client client.Client,
 	loggingDataProvider loggingdataprovider.LoggingDataProvider,
 	nameProvider NameProvider) *Reconciler {
 	return &Reconciler{
-		Logging:                   logging,
-		GenericResourceReconciler: reconciler.NewGenericReconciler(client, logger, opts),
-		fluentbitSpec:             fluentbitSpec,
-		loggingDataProvider:       loggingDataProvider,
-		nameProvider:              nameProvider,
+		Logging:             logging,
+		logger:              logger,
+		resourceReconciler:  reconciler.NewGenericReconciler(client, logger.WithName("reconciler"), opts),
+		fluentbitSpec:       fluentbitSpec,
+		loggingDataProvider: loggingDataProvider,
+		nameProvider:        nameProvider,
 	}
 }
 
@@ -129,7 +140,7 @@ func (r *Reconciler) Reconcile() (*reconcile.Result, error) {
 		if o == nil {
 			return nil, errors.Errorf("Reconcile error! Resource %#v returns with nil object", factory)
 		}
-		result, err := r.ReconcileResource(o, state)
+		result, err := r.resourceReconciler.ReconcileResource(o, state)
 		if err != nil {
 			return nil, errors.WrapWithDetails(err,
 				"failed to reconcile resource", "resource", o.GetObjectKind().GroupVersionKind())

pkg/resources/fluentbit/psp.go

@@ -53,7 +53,7 @@ func (r *Reconciler) clusterPodSecurityPolicy() (runtime.Object, reconciler.Desi
 
 		if r.fluentbitSpec.PositionDB.HostPath != nil {
 			r.fluentbitSpec.PositionDB.WithDefaultHostPath(
-				fmt.Sprintf(v1beta1.HostPath, r.Logging.Name, TailPositionVolume))
+				fmt.Sprintf(v1beta1.HostPath, r.nameProvider.Name(), TailPositionVolume))
 
 			allowedHostPaths = append(allowedHostPaths, policyv1beta1.AllowedHostPath{
 				PathPrefix: r.fluentbitSpec.PositionDB.HostPath.Path,

pkg/resources/model/repository.go

@@ -21,9 +21,10 @@ import (
 
 	"emperror.dev/errors"
 	"github.com/go-logr/logr"
-	"github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1"
 	corev1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1"
 )
 
 func NewLoggingResourceRepository(client client.Reader, logger logr.Logger) *LoggingResourceRepository {

pkg/sdk/logging/api/v1beta1/fluentbit_types.go

@@ -472,6 +472,13 @@ func (l *FluentbitNameProvider) ComponentName(name string) string {
 	return fmt.Sprintf("%s-%s", l.fluentbit.Name, name)
 }
 
+func (l *FluentbitNameProvider) Name() string {
+	if l.logging != nil {
+		return l.logging.Name
+	}
+	return l.fluentbit.Name
+}
+
 func (l *FluentbitNameProvider) OwnerRef() metav1.OwnerReference {
 	if l.logging != nil {
 		return metav1.OwnerReference{