Cherry-pick #18630 to 7.x: Validate config before init NewInput to avoid uneccessory cleanup and kubernetes watcher leak

CHANGELOG.next.asciidoc

@@ -173,6 +173,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix a rate limit related issue in httpjson input for Okta module. {issue}18530[18530] {pull}18534[18534]
 - Fixed ingestion of some Cisco ASA and FTD messages when a hostname was used instead of an IP for NAT fields. {issue}14034[14034] {pull}18376[18376]
 - Fix `o365.audit` failing to ingest events when ip address is surrounded by square brackets. {issue}18587[18587] {pull}18591[18591]
+- Fix Kubernetes Watcher goroutine leaks when input config is invalid and `input.reload` is enabled. {issue}18629[18629] {pull}18630[18630]
 
 *Heartbeat*
 

filebeat/input/log/input.go

@@ -85,6 +85,22 @@ func NewInput(
 		}
 	}
 
+	inputConfig := defaultConfig
+
+	if err := cfg.Unpack(&inputConfig); err != nil {
+		return nil, err
+	}
+	if err := inputConfig.resolveRecursiveGlobs(); err != nil {
+		return nil, fmt.Errorf("Failed to resolve recursive globs in config: %v", err)
+	}
+	if err := inputConfig.normalizeGlobPatterns(); err != nil {
+		return nil, fmt.Errorf("Failed to normalize globs patterns: %v", err)
+	}
+
+	if len(inputConfig.Paths) == 0 {
+		return nil, fmt.Errorf("each input must have at least one path defined")
+	}
+
 	// Note: underlying output.
 	//  The input and harvester do have different requirements
 	//  on the timings the outlets must be closed/unblocked.
@@ -113,7 +129,7 @@ func NewInput(
 	}
 
 	p := &Input{
-		config:      defaultConfig,
+		config:      inputConfig,
 		cfg:         cfg,
 		harvesters:  harvester.NewRegistry(),
 		outlet:      out,
@@ -123,27 +139,13 @@ func NewInput(
 		meta:        meta,
 	}
 
-	if err := cfg.Unpack(&p.config); err != nil {
-		return nil, err
-	}
-	if err := p.config.resolveRecursiveGlobs(); err != nil {
-		return nil, fmt.Errorf("Failed to resolve recursive globs in config: %v", err)
-	}
-	if err := p.config.normalizeGlobPatterns(); err != nil {
-		return nil, fmt.Errorf("Failed to normalize globs patterns: %v", err)
-	}
-
 	// Create empty harvester to check if configs are fine
 	// TODO: Do config validation instead
 	_, err = p.createHarvester(file.State{}, nil)
 	if err != nil {
 		return nil, err
 	}
 
-	if len(p.config.Paths) == 0 {
-		return nil, fmt.Errorf("each input must have at least one path defined")
-	}
-
 	err = p.loadStates(context.States)
 	if err != nil {
 		return nil, err

libbeat/common/kubernetes/util.go

@@ -97,7 +97,7 @@ func DiscoverKubernetesNode(log *logp.Logger, host string, inCluster bool, clien
 			log.Errorf("kubernetes: Querying for pod failed with error: %+v", err)
 			return defaultNode
 		}
-		log.Info("kubernetes: Using node %s discovered by in cluster pod node query", pod.Spec.NodeName)
+		log.Infof("kubernetes: Using node %s discovered by in cluster pod node query", pod.Spec.NodeName)
 		return pod.Spec.NodeName
 	}