[Chore][libbeat] Replace global logger with single logger instance (#43493)

* [Chore] Use single logger instance

* Update x-pack/filebeat/fbreceiver/receiver_test.go

Co-authored-by: Mauri de Souza Meneguzzo <[email protected]>

* Update x-pack/filebeat/fbreceiver/receiver_test.go

Co-authored-by: Mauri de Souza Meneguzzo <[email protected]>

* libbeat/cmd is complete

* keep only libbeat changes

* keep only libbeat changes

* add libbeat/monitoring/report package

* add my fork

* use beat.Info.Logger

* remove cloudid

* remove println

* fix tests

* add libbeat/idxmgmt

* add libbeat/monitoring/report

* go mod tidy

* fix idxmgmt tests

* fixing and finishing things

* fix tests

* fix tests

* fix tests

* debugging statement

* stringify debug statement

* fixing tests

* fixing tests

* add libbeat/instrumentation

* update ea-libs

* fix lint

* fix tests and lint

* golangci lint

* initialize logger before using in instrumentation

* address review comments

* address review comments 2

* add libbeat/output & libbeat/template

* fix tests

* add libbeat/dashboards & libbeat/cfgfile

* add libbeat/dashboards & libbeat/cfgfile

* fix lint

* add libbeat/persistentcache

* mage check

* mage check

* merge and fix tests

* fix test

* add logger on metricbeat

* fix lint

* fix lint

* fix tests

* address review comments

* lint check

* fix lint

* fix lint

* fix lint

* fix lint

* fix lint

* fix lint in redis files

* lint fix

* fix lint

* fix lint and tests

* fix logstash test

* fix test reload functionality

* add integration build tag

* use t.TempDir and non-capitalize logged errors

* add linter check

* fix log discrepancy

* add nolint for errors starting with capitalized letter

* fix logger discrepancy

* add linting fixes

---------

Co-authored-by: Mauri de Souza Meneguzzo <[email protected]>

Author: khushijain21

filebeat/beater/crawler.go

@@ -48,9 +48,10 @@ func newCrawler(
 	inputConfigs []*conf.C,
 	beatDone chan struct{},
 	once bool,
+	logger *logp.Logger,
 ) (*crawler, error) {
 	return &crawler{
-		log:            logp.NewLogger("crawler"),
+		log:            logger.Named("crawler"),
 		inputs:         map[uint64]cfgfile.Runner{},
 		inputsFactory:  inputFactory,
 		modulesFactory: module,
@@ -152,7 +153,7 @@ func (c *crawler) startInput(
 }
 
 func (c *crawler) Stop() {
-	logp.Info("Stopping Crawler")
+	c.log.Info("Stopping Crawler")
 
 	asyncWaitStop := func(stop func()) {
 		c.wg.Add(1)
@@ -162,7 +163,7 @@ func (c *crawler) Stop() {
 		}()
 	}
 
-	logp.Info("Stopping %d inputs", len(c.inputs))
+	c.log.Infof("Stopping %d inputs", len(c.inputs))
 	// Stop inputs in parallel
 	for id, p := range c.inputs {
 		id, p := id, p
@@ -182,7 +183,7 @@ func (c *crawler) Stop() {
 
 	c.WaitForCompletion()
 
-	logp.Info("Crawler stopped")
+	c.log.Info("Crawler stopped")
 }
 
 func (c *crawler) WaitForCompletion() {

filebeat/beater/filebeat.go

@@ -73,6 +73,7 @@ type Filebeat struct {
 	done           chan struct{}
 	stopOnce       sync.Once // wraps the Stop() method
 	pipeline       beat.PipelineConnector
+	logger         *logp.Logger
 }
 
 type PluginFactory func(beat.Info, *logp.Logger, statestore.States) []v2.Plugin
@@ -87,7 +88,7 @@ func New(plugins PluginFactory) beat.Creator {
 func newBeater(b *beat.Beat, plugins PluginFactory, rawConfig *conf.C) (beat.Beater, error) {
 	config := cfg.DefaultConfig
 	if err := rawConfig.Unpack(&config); err != nil {
-		return nil, fmt.Errorf("Error reading config file: %w", err)
+		return nil, fmt.Errorf("Error reading config file: %w", err) //nolint:staticcheck //Keep old behavior
 	}
 
 	if err := cfgwarn.CheckRemoved6xSettings(
@@ -132,7 +133,7 @@ func newBeater(b *beat.Beat, plugins PluginFactory, rawConfig *conf.C) (beat.Bea
 			"input_metrics.json", "application/json", func() []byte {
 				data, err := inputmon.MetricSnapshotJSON()
 				if err != nil {
-					logp.L().Warnw("Failed to collect input metric snapshot for Agent diagnostics.", "error", err)
+					b.Info.Logger.Warnw("Failed to collect input metric snapshot for Agent diagnostics.", "error", err)
 					return []byte(err.Error())
 				}
 				return data
@@ -161,7 +162,7 @@ func newBeater(b *beat.Beat, plugins PluginFactory, rawConfig *conf.C) (beat.Bea
 		}
 
 		// in the `setup` command, log this only as a warning
-		logp.Warn("Setup called, but no modules enabled.")
+		b.Info.Logger.Warn("Setup called, but no modules enabled.")
 	}
 
 	if *once && config.ConfigInput.Enabled() && config.ConfigModules.Enabled() {
@@ -177,6 +178,7 @@ func newBeater(b *beat.Beat, plugins PluginFactory, rawConfig *conf.C) (beat.Bea
 		config:         &config,
 		moduleRegistry: moduleRegistry,
 		pluginFactory:  plugins,
+		logger:         b.Info.Logger,
 	}
 
 	err = fb.setupPipelineLoaderCallback(b)
@@ -190,7 +192,7 @@ func newBeater(b *beat.Beat, plugins PluginFactory, rawConfig *conf.C) (beat.Bea
 // setupPipelineLoaderCallback sets the callback function for loading pipelines during setup.
 func (fb *Filebeat) setupPipelineLoaderCallback(b *beat.Beat) error {
 	if b.Config.Output.Name() != "elasticsearch" && !b.Manager.Enabled() {
-		logp.Warn(pipelinesWarning)
+		fb.logger.Warn(pipelinesWarning)
 		return nil
 	}
 
@@ -223,7 +225,7 @@ func (fb *Filebeat) setupPipelineLoaderCallback(b *beat.Beat) error {
 				newPath := strings.TrimSuffix(origPath, ".yml")
 				_ = fb.config.ConfigModules.SetString("path", -1, newPath)
 			}
-			modulesLoader := cfgfile.NewReloader(logp.L().Named("module.reloader"), fb.pipeline, fb.config.ConfigModules)
+			modulesLoader := cfgfile.NewReloader(fb.logger.Named("module.reloader"), fb.pipeline, fb.config.ConfigModules)
 			modulesLoader.Load(modulesFactory)
 		}
 
@@ -236,7 +238,7 @@ func (fb *Filebeat) setupPipelineLoaderCallback(b *beat.Beat) error {
 // setup.
 func (fb *Filebeat) loadModulesPipelines(b *beat.Beat) error {
 	if b.Config.Output.Name() != "elasticsearch" {
-		logp.Warn(pipelinesWarning)
+		fb.logger.Warn(pipelinesWarning)
 		return nil
 	}
 
@@ -288,7 +290,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 
 	registryMigrator := registrar.NewMigrator(config.Registry)
 	if err := registryMigrator.Run(); err != nil {
-		logp.Err("Failed to migrate registry file: %+v", err)
+		fb.logger.Errorf("Failed to migrate registry file: %+v", err)
 		return err
 	}
 
@@ -299,9 +301,9 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 		cn()
 	}()
 
-	stateStore, err := openStateStore(ctx, b.Info, logp.NewLogger("filebeat"), config.Registry)
+	stateStore, err := openStateStore(ctx, b.Info, fb.logger.Named("filebeat"), config.Registry)
 	if err != nil {
-		logp.Err("Failed to open state store: %+v", err)
+		fb.logger.Errorf("Failed to open state store: %+v", err)
 		return err
 	}
 	defer stateStore.Close()
@@ -313,7 +315,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 		b.OutputConfigReloader = reload.ReloadableFunc(func(r *reload.ConfigWithMeta) error {
 			outCfg := conf.Namespace{}
 			if err := r.Config.Unpack(&outCfg); err != nil || outCfg.Name() != "elasticsearch" {
-				logp.Err("Failed to unpack the output config: %v", err)
+				fb.logger.Errorf("Failed to unpack the output config: %v", err)
 				return nil
 			}
 
@@ -322,24 +324,24 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 			// See https://github.com/elastic/beats/issues/42815
 			configCopy, err := conf.NewConfigFrom(outCfg.Config())
 			if err != nil {
-				logp.Err("Failed to create a new config from the output config: %v", err)
+				fb.logger.Errorf("Failed to create a new config from the output config: %v", err)
 				return nil
 			}
 			stateStore.notifier.Notify(configCopy)
 			return nil
 		})
 	}
 
-	err = filestream.ValidateInputIDs(config.Inputs, logp.NewLogger("input.filestream"))
+	err = filestream.ValidateInputIDs(config.Inputs, fb.logger.Named("input.filestream"))
 	if err != nil {
-		logp.Err("invalid filestream configuration: %+v", err)
+		fb.logger.Errorf("invalid filestream configuration: %+v", err)
 		return err
 	}
 
 	// Setup registrar to persist state
 	registrar, err := registrar.New(stateStore, finishedLogger, config.Registry.FlushTimeout)
 	if err != nil {
-		logp.Err("Could not init registrar: %v", err)
+		fb.logger.Errorf("Could not init registrar: %v", err)
 		return err
 	}
 
@@ -364,7 +366,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 	outDone := make(chan struct{}) // outDone closes down all active pipeline connections
 	pipelineConnector := channel.NewOutletFactory(outDone).Create
 
-	inputsLogger := logp.NewLogger("input")
+	inputsLogger := fb.logger.Named("input")
 	v2Inputs := fb.pluginFactory(b.Info, inputsLogger, stateStore)
 	v2InputLoader, err := v2.NewLoader(inputsLogger, v2Inputs, "type", cfg.DefaultType)
 	if err != nil {
@@ -378,7 +380,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 
 	// Store needs to be fully configured at this point
 	if err := v2InputLoader.Init(&inputTaskGroup); err != nil {
-		logp.Err("Failed to initialize the input managers: %v", err)
+		fb.logger.Errorf("Failed to initialize the input managers: %v", err)
 		return err
 	}
 
@@ -399,12 +401,12 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 	if b.Config.Output.Name() == "elasticsearch" {
 		pipelineLoaderFactory = newPipelineLoaderFactory(pipelineFactoryCtx, b.Config.Output.Config())
 	} else {
-		logp.Warn(pipelinesWarning)
+		fb.logger.Warn(pipelinesWarning)
 	}
 	moduleLoader := fileset.NewFactory(inputLoader, b.Info, pipelineLoaderFactory, config.OverwritePipelines)
-	crawler, err := newCrawler(inputLoader, moduleLoader, config.Inputs, fb.done, *once)
+	crawler, err := newCrawler(inputLoader, moduleLoader, config.Inputs, fb.done, *once, fb.logger)
 	if err != nil {
-		logp.Err("Could not init crawler: %v", err)
+		fb.logger.Errorf("Could not init crawler: %v", err)
 		return err
 	}
 
@@ -415,7 +417,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 	// Start the registrar
 	err = registrar.Start()
 	if err != nil {
-		return fmt.Errorf("Could not start registrar: %w", err)
+		return fmt.Errorf("Could not start registrar: %w", err) //nolint:staticcheck //Keep old behavior
 	}
 
 	// Stopping registrar will write last state
@@ -433,31 +435,31 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 	defer waitEvents.Wait()
 
 	if config.OverwritePipelines {
-		logp.Debug("modules", "Existing Ingest pipelines will be updated")
+		fb.logger.Debug("modules", "Existing Ingest pipelines will be updated")
 	}
 
 	err = crawler.Start(fb.pipeline, config.ConfigInput, config.ConfigModules)
 	if err != nil {
 		crawler.Stop()
 		cancelPipelineFactoryCtx()
-		return fmt.Errorf("Failed to start crawler: %w", err)
+		return fmt.Errorf("Failed to start crawler: %w", err) //nolint:staticcheck //Keep old behavior
 	}
 
 	// If run once, add crawler completion check as alternative to done signal
 	if *once {
 		runOnce := func() {
-			logp.Info("Running filebeat once. Waiting for completion ...")
+			fb.logger.Info("Running filebeat once. Waiting for completion ...")
 			crawler.WaitForCompletion()
-			logp.Info("All data collection completed. Shutting down.")
+			fb.logger.Info("All data collection completed. Shutting down.")
 		}
 		waitFinished.Add(runOnce)
 	}
 
 	// Register reloadable list of inputs and modules
-	inputs := cfgfile.NewRunnerList(management.DebugK, inputLoader, fb.pipeline)
+	inputs := cfgfile.NewRunnerList(management.DebugK, inputLoader, fb.pipeline, fb.logger)
 	b.Registry.MustRegisterInput(inputs)
 
-	modules := cfgfile.NewRunnerList(management.DebugK, moduleLoader, fb.pipeline)
+	modules := cfgfile.NewRunnerList(management.DebugK, moduleLoader, fb.pipeline, fb.logger)
 
 	var adiscover *autodiscover.Autodiscover
 	if fb.config.Autodiscover != nil {
@@ -471,6 +473,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 			autodiscover.QueryConfig(),
 			config.Autodiscover,
 			b.Keystore,
+			fb.logger,
 		)
 		if err != nil {
 			return err
@@ -506,7 +509,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 			"Continue shutdown: All enqueued events being published."))
 		// Wait for either timeout or all events having been ACKed by outputs.
 		if fb.config.ShutdownTimeout > 0 {
-			logp.Info("Shutdown output timer started. Waiting for max %v.", timeout)
+			fb.logger.Info("Shutdown output timer started. Waiting for max %v.", timeout)
 			waitEvents.Add(withLog(waitDuration(timeout),
 				"Continue shutdown: Time out waiting for events being published."))
 		} else {
@@ -532,7 +535,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 
 // Stop is called on exit to stop the crawling, spooling and registration processes.
 func (fb *Filebeat) Stop() {
-	logp.Info("Stopping filebeat")
+	fb.logger.Info("Stopping filebeat")
 
 	// Stop Filebeat
 	fb.stopOnce.Do(func() { close(fb.done) })
@@ -543,7 +546,7 @@ func newPipelineLoaderFactory(ctx context.Context, esConfig *conf.C) fileset.Pip
 	pipelineLoaderFactory := func() (fileset.PipelineLoader, error) {
 		esClient, err := eslegclient.NewConnectedClient(ctx, esConfig, "Filebeat")
 		if err != nil {
-			return nil, fmt.Errorf("Error creating Elasticsearch client: %w", err)
+			return nil, fmt.Errorf("Error creating Elasticsearch client: %w", err) //nolint:staticcheck //Keep old behavior
 		}
 		return esClient, nil
 	}

filebeat/cmd/modules.go

@@ -39,7 +39,7 @@ func buildModulesManager(beat *beat.Beat) (cmd.ModulesManager, error) {
 		return nil, fmt.Errorf("wrong settings for config.modules.path, it is expected to end with *.yml. Got: %s", glob)
 	}
 
-	modulesManager, err := cfgfile.NewGlobManager(glob, ".yml", ".disabled")
+	modulesManager, err := cfgfile.NewGlobManager(glob, ".yml", ".disabled", beat.Info.Logger)
 	if err != nil {
 		return nil, fmt.Errorf("initialization error: %w", err)
 	}

heartbeat/beater/heartbeat.go

@@ -184,7 +184,7 @@ func (bt *Heartbeat) Run(b *beat.Beat) error {
 	}
 
 	if bt.config.ConfigMonitors.Enabled() {
-		bt.monitorReloader = cfgfile.NewReloader(logp.L().Named("module.reload"), b.Publisher, bt.config.ConfigMonitors)
+		bt.monitorReloader = cfgfile.NewReloader(b.Info.Logger.Named("module.reload"), b.Publisher, bt.config.ConfigMonitors)
 		defer bt.monitorReloader.Stop()
 
 		err := bt.RunReloadableMonitors()
@@ -285,7 +285,7 @@ func (bt *Heartbeat) RunCentralMgmtMonitors(b *beat.Beat) {
 		return nil
 	})
 
-	inputs := cfgfile.NewRunnerList(management.DebugK, bt.monitorFactory, b.Publisher)
+	inputs := cfgfile.NewRunnerList(management.DebugK, bt.monitorFactory, b.Publisher, b.Info.Logger)
 	b.Registry.MustRegisterInput(inputs)
 }
 
@@ -311,6 +311,7 @@ func (bt *Heartbeat) makeAutodiscover(b *beat.Beat) (*autodiscover.Autodiscover,
 		autodiscover.QueryConfig(),
 		bt.config.Autodiscover,
 		b.Keystore,
+		b.Info.Logger,
 	)
 	if err != nil {
 		return nil, err

libbeat/autodiscover/autodiscover.go

@@ -75,8 +75,9 @@ func NewAutodiscover(
 	configurer EventConfigurer,
 	c *Config,
 	keystore keystore.Keystore,
+	logger *logp.Logger,
 ) (*Autodiscover, error) {
-	logger := logp.NewLogger("autodiscover")
+	logger = logger.Named("autodiscover")
 
 	// Init Event bus
 	bus := bus.New(logger, name)
@@ -98,7 +99,7 @@ func NewAutodiscover(
 		factory:         factory,
 		configurer:      configurer,
 		configs:         map[string]map[uint64]*reload.ConfigWithMeta{},
-		runners:         cfgfile.NewRunnerList("autodiscover.cfgfile", factory, pipeline),
+		runners:         cfgfile.NewRunnerList("autodiscover.cfgfile", factory, pipeline, logger),
 		providers:       providers,
 		meta:            meta.NewMap(),
 		logger:          logger,