filebeat: rename source log key to source_file (#30688)

Filebeat's logs were not fully compatible with ECS because some log entries were using `source` as string. This commit fixes it by rename `source` to `source_file`. Fixes: #30667

Author: belimawr

CHANGELOG.next.asciidoc

@@ -58,6 +58,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...main[Check the HEAD dif
 - elasticsearch: fix duplicate ingest when using a common appender configuration {issue}30428[30428] {pull}30440[30440]
 - Fix ECS version string in threatintel to be consistent with other modules and add event.timezone. {issue}30499[30499] {pull}30570[30570]
 - Add default paths value to MySQL Enterprise module to prevent issues with pipeline installations {pull}30598[30598]
+- Fix compatibility with ECS by renaming `source` log key to `source_file` {issue}30667[30667]
 
 *Heartbeat*
 

filebeat/input/filestream/internal/input-logfile/harvester.go

@@ -141,7 +141,7 @@ type defaultHarvesterGroup struct {
 func (hg *defaultHarvesterGroup) Start(ctx input.Context, s Source) {
 sourceName := hg.identifier.ID(s)
 
-ctx.Logger = ctx.Logger.With("source", sourceName)
+ctx.Logger = ctx.Logger.With("source_file", sourceName)
 ctx.Logger.Debug("Starting harvester for file")
 
 hg.tg.Go(startHarvester(ctx, hg, s, false))
@@ -152,7 +152,7 @@ func (hg *defaultHarvesterGroup) Start(ctx input.Context, s Source) {
 func (hg *defaultHarvesterGroup) Restart(ctx input.Context, s Source) {
 sourceName := hg.identifier.ID(s)
 
-ctx.Logger = ctx.Logger.With("source", sourceName)
+ctx.Logger = ctx.Logger.With("source_file", sourceName)
 ctx.Logger.Debug("Restarting harvester for file")
 
 hg.tg.Go(startHarvester(ctx, hg, s, true))

filebeat/input/log/logger.go

@@ -24,7 +24,7 @@ import (
 
 func loggerWithState(logger *logp.Logger, state file.State) *logp.Logger {
 return logger.With(
-"source", state.Source,
+"source_file", state.Source,
 "state_id", state.Id,
 "finished", state.Finished,
 "os_id", state.FileStateOS,