Merge pull request #1665 from kube-logging/fix-fluentbit-tenants-detached-aggregator

fix to make detached aggregators work when generating fluentbit tenants

Author: pepov

controllers/logging/logging_controller.go

@@ -218,13 +218,11 @@ func (r *LoggingReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 				r.Client,
 				log.WithName("fluentbit-legacy"),
 				&logging,
-				fluentdSpec,
-				syslogNGSpec,
 				reconcilerOpts,
 				logging.Spec.FluentbitSpec,
 				loggingDataProvider,
 				nameProvider,
-				nil, // Not implemented for the embedded fluentbit config
+				loggingResourceRepo,
 			).Reconcile)
 		}
 	default:
@@ -238,13 +236,11 @@ func (r *LoggingReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 				r.Client,
 				l.WithValues("fluentbitagent", f.Name),
 				&logging,
-				fluentdSpec,
-				syslogNGSpec,
 				reconcilerOpts,
 				&f.Spec,
 				loggingDataProvider,
 				fluentbit.NewStandaloneFluentbitNameProvider(&f),
-				loggingResources.LoggingRoutes,
+				loggingResourceRepo,
 			).Reconcile)
 		}
 	}

pkg/resources/fluentbit/configsecret.go

@@ -238,7 +238,14 @@ func (r *Reconciler) configSecret() (runtime.Object, reconciler.DesiredState, er
 		}
 	}
 
-	if r.fluentdSpec != nil {
+	loggingResources, errs := r.loggingResourcesRepo.LoggingResourcesFor(ctx, *r.Logging)
+	if errs != nil {
+		return nil, nil, errs
+	}
+
+	fluentdSpec := loggingResources.GetFluentdSpec()
+
+	if fluentdSpec != nil {
 		fluentbitTargetHost := r.fluentbitSpec.TargetHost
 		if fluentbitTargetHost == "" {
 			fluentbitTargetHost = aggregatorEndpoint(r.Logging, fluentd.ServiceName)
@@ -354,15 +361,15 @@ func (r *Reconciler) configSecret() (runtime.Object, reconciler.DesiredState, er
 		}
 	}
 
-	if r.syslogNGSpec != nil {
+	if loggingResources.GetSyslogNGSpec() != nil {
 		input.SyslogNGOutput = newSyslogNGOutputConfig()
 		input.SyslogNGOutput.Host = aggregatorEndpoint(r.Logging, syslogng.ServiceName)
 		input.SyslogNGOutput.Port = syslogng.ServicePort
 	}
 
-	if len(r.loggingRoutes) > 0 {
+	if len(loggingResources.LoggingRoutes) > 0 {
 		var tenants []v1beta1.Tenant
-		for _, a := range r.loggingRoutes {
+		for _, a := range loggingResources.LoggingRoutes {
 			tenants = append(tenants, a.Status.Tenants...)
 		}
 		if err := r.configureOutputsForTenants(ctx, tenants, &input); err != nil {

pkg/resources/fluentbit/fluentbit.go

@@ -23,6 +23,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 
 	"github.com/kube-logging/logging-operator/pkg/resources/loggingdataprovider"
+	"github.com/kube-logging/logging-operator/pkg/resources/model"
 
 	"github.com/cisco-open/operator-tools/pkg/reconciler"
 	util "github.com/cisco-open/operator-tools/pkg/utils"
@@ -87,39 +88,33 @@ type DesiredObject struct {
 
 // Reconciler holds info what resource to reconcile
 type Reconciler struct {
-	resourceReconciler  *reconciler.GenericResourceReconciler
-	logger              logr.Logger
-	Logging             *v1beta1.Logging
-	fluentdSpec         *v1beta1.FluentdSpec
-	syslogNGSpec        *v1beta1.SyslogNGSpec
-	configs             map[string][]byte
-	fluentbitSpec       *v1beta1.FluentbitSpec
-	loggingDataProvider loggingdataprovider.LoggingDataProvider
-	nameProvider        NameProvider
-	loggingRoutes       []v1beta1.LoggingRoute
+	resourceReconciler   *reconciler.GenericResourceReconciler
+	logger               logr.Logger
+	Logging              *v1beta1.Logging
+	configs              map[string][]byte
+	fluentbitSpec        *v1beta1.FluentbitSpec
+	loggingDataProvider  loggingdataprovider.LoggingDataProvider
+	nameProvider         NameProvider
+	loggingResourcesRepo *model.LoggingResourceRepository
 }
 
 // NewReconciler creates a new FluentbitAgent reconciler
 func New(client client.Client,
 	logger logr.Logger,
 	logging *v1beta1.Logging,
-	fluentdSpec *v1beta1.FluentdSpec,
-	syslogNGSpec *v1beta1.SyslogNGSpec,
 	opts reconciler.ReconcilerOpts,
 	fluentbitSpec *v1beta1.FluentbitSpec,
 	loggingDataProvider loggingdataprovider.LoggingDataProvider,
 	nameProvider NameProvider,
-	loggingRoutes []v1beta1.LoggingRoute) *Reconciler {
+	loggingResourcesRepo *model.LoggingResourceRepository) *Reconciler {
 	return &Reconciler{
-		Logging:             logging,
-		fluentdSpec:         fluentdSpec,
-		syslogNGSpec:        syslogNGSpec,
-		logger:              logger,
-		resourceReconciler:  reconciler.NewGenericReconciler(client, logger.WithName("reconciler"), opts),
-		fluentbitSpec:       fluentbitSpec,
-		loggingDataProvider: loggingDataProvider,
-		nameProvider:        nameProvider,
-		loggingRoutes:       loggingRoutes,
+		Logging:              logging,
+		logger:               logger,
+		resourceReconciler:   reconciler.NewGenericReconciler(client, logger.WithName("reconciler"), opts),
+		fluentbitSpec:        fluentbitSpec,
+		loggingDataProvider:  loggingDataProvider,
+		nameProvider:         nameProvider,
+		loggingResourcesRepo: loggingResourcesRepo,
 	}
 }
 

pkg/resources/fluentbit/tenants.go

@@ -100,7 +100,14 @@ func (r *Reconciler) configureOutputsForTenants(ctx context.Context, tenants []v
 		if err := r.resourceReconciler.Client.Get(ctx, types.NamespacedName{Name: t.Name}, logging); err != nil {
 			return errors.WrapIf(err, "getting logging resource")
 		}
-		if logging.Spec.FluentdSpec != nil {
+
+		loggingResources, err := r.loggingResourcesRepo.LoggingResourcesFor(ctx, *logging)
+		if err != nil {
+			errs = errors.Append(errs, errors.WrapIff(err, "querying related resources for logging %s", logging.Name))
+			continue
+		}
+
+		if loggingResources.GetFluentdSpec() != nil {
 			if input.FluentForwardOutput == nil {
 				input.FluentForwardOutput = &fluentForwardOutputConfig{}
 			}
@@ -110,7 +117,7 @@ func (r *Reconciler) configureOutputsForTenants(ctx context.Context, tenants []v
 				Host:           aggregatorEndpoint(logging, fluentd.ServiceName),
 				Port:           fluentd.ServicePort,
 			})
-		} else if logging.Spec.SyslogNGSpec != nil {
+		} else if loggingResources.GetSyslogNGSpec() != nil {
 			if input.SyslogNGOutput == nil {
 				input.SyslogNGOutput = newSyslogNGOutputConfig()
 			}

pkg/resources/model/resources.go

@@ -29,7 +29,7 @@ type LoggingResources struct {
 	WatchNamespaces []string
 }
 
-func (l LoggingResources) GetFluentd() *v1beta1.FluentdConfig {
+func (l LoggingResources) getFluentd() *v1beta1.FluentdConfig {
 	if l.Fluentd.Configuration != nil {
 		return l.Fluentd.Configuration
 	}
@@ -38,7 +38,7 @@ func (l LoggingResources) GetFluentd() *v1beta1.FluentdConfig {
 
 func (l LoggingResources) GetFluentdSpec() *v1beta1.FluentdSpec {
 
-	if detachedFluentd := l.GetFluentd(); detachedFluentd != nil {
+	if detachedFluentd := l.getFluentd(); detachedFluentd != nil {
 		return &detachedFluentd.Spec
 	}
 	if l.Logging.Spec.FluentdSpec != nil {
@@ -57,7 +57,7 @@ type FluentdLoggingResources struct {
 	ExcessFluentds []v1beta1.FluentdConfig
 }
 
-func (l LoggingResources) GetSyslogNG() *v1beta1.SyslogNGConfig {
+func (l LoggingResources) getSyslogNG() *v1beta1.SyslogNGConfig {
 	if l.SyslogNG.Configuration != nil {
 		return l.SyslogNG.Configuration
 	}
@@ -66,7 +66,7 @@ func (l LoggingResources) GetSyslogNG() *v1beta1.SyslogNGConfig {
 
 func (l LoggingResources) GetSyslogNGSpec() *v1beta1.SyslogNGSpec {
 
-	if detachedSyslogNG := l.GetSyslogNG(); detachedSyslogNG != nil {
+	if detachedSyslogNG := l.getSyslogNG(); detachedSyslogNG != nil {
 		return &detachedSyslogNG.Spec
 	}
 	if l.Logging.Spec.SyslogNGSpec != nil {