Fix leak caused by input runners created when checking their configuration (#23722)

Stop input v1 runners created to check config.

`CheckConfig` for v1 inputs actually calls the constructors of the inputs.
In some cases, as in the log input, the constructor creates resources that
are never released unless the runner is stopped. This causes goroutines
leaks with autodiscover or other dynamic configurations.

Author: jsoriano

CHANGELOG.next.asciidoc

@@ -280,6 +280,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Change the `event.created` in Netflow events to be the time the event was created by Filebeat
   to be consistent with ECS. {pull}23094[23094]
 - Update `filestream` reader offset when a line is skipped. {pull}23417[23417]
+- Fix goroutines leak with some inputs in autodiscover. {pull}23722[23722]
 
 *Filebeat*
 

filebeat/fileset/factory.go

@@ -18,7 +18,10 @@
 package fileset
 
 import (
+	"fmt"
+
 	"github.com/gofrs/uuid"
+	"github.com/mitchellh/hashstructure"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/cfgfile"
@@ -27,9 +30,6 @@ import (
 	"github.com/elastic/beats/v7/libbeat/logp"
 	"github.com/elastic/beats/v7/libbeat/monitoring"
 	"github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
-	pubpipeline "github.com/elastic/beats/v7/libbeat/publisher/pipeline"
-
-	"github.com/mitchellh/hashstructure"
 )
 
 var (
@@ -77,15 +77,9 @@ func NewFactory(
 
 // Create creates a module based on a config
 func (f *Factory) Create(p beat.PipelineConnector, c *common.Config) (cfgfile.Runner, error) {
-	// Start a registry of one module:
-	m, err := NewModuleRegistry([]*common.Config{c}, f.beatInfo, false)
-	if err != nil {
-		return nil, err
-	}
-
-	pConfigs, err := m.GetInputConfigs()
+	m, pConfigs, err := f.createRegistry(c)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("could not create module registry for filesets: %w", err)
 	}
 
 	// Hash module ID
@@ -116,8 +110,36 @@ func (f *Factory) Create(p beat.PipelineConnector, c *common.Config) (cfgfile.Ru
 }
 
 func (f *Factory) CheckConfig(c *common.Config) error {
-	_, err := f.Create(pubpipeline.NewNilPipeline(), c)
-	return err
+	_, pConfigs, err := f.createRegistry(c)
+	if err != nil {
+		return fmt.Errorf("could not create module registry for filesets: %w", err)
+	}
+
+	for _, pConfig := range pConfigs {
+		err = f.inputFactory.CheckConfig(pConfig)
+		if err != nil {
+			logp.Err("Error checking input configuration: %s", err)
+			return err
+		}
+	}
+
+	return nil
+}
+
+// createRegistry starts a registry for a set of filesets, it returns the registry and
+// its input configurations
+func (f *Factory) createRegistry(c *common.Config) (*ModuleRegistry, []*common.Config, error) {
+	m, err := NewModuleRegistry([]*common.Config{c}, f.beatInfo, false)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	pConfigs, err := m.GetInputConfigs()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return m, pConfigs, err
 }
 
 func (p *inputsRunner) Start() {

filebeat/input/container/input_test.go

@@ -0,0 +1,36 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// +build !integration
+
+package container
+
+import (
+	"os"
+	"path"
+	"testing"
+
+	"github.com/elastic/beats/v7/filebeat/input/inputtest"
+	"github.com/elastic/beats/v7/libbeat/common"
+)
+
+func TestNewInputDone(t *testing.T) {
+	config := common.MapStr{
+		"paths": path.Join(os.TempDir(), "logs", "*.log"),
+	}
+	inputtest.AssertNotStartedInputCanBeDone(t, NewInput, &config)
+}

filebeat/input/docker/input_test.go

@@ -0,0 +1,34 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// +build !integration
+
+package docker
+
+import (
+	"testing"
+
+	"github.com/elastic/beats/v7/filebeat/input/inputtest"
+	"github.com/elastic/beats/v7/libbeat/common"
+)
+
+func TestNewInputDone(t *testing.T) {
+	config := common.MapStr{
+		"containers.ids": "fad130edd3d2",
+	}
+	inputtest.AssertNotStartedInputCanBeDone(t, NewInput, &config)
+}

filebeat/input/inputtest/input.go

@@ -0,0 +1,65 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package inputtest
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/beats/v7/filebeat/channel"
+	"github.com/elastic/beats/v7/filebeat/input"
+	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/tests/resources"
+)
+
+// Outlet is an empty outlet for testing.
+type Outlet struct{}
+
+func (o Outlet) OnEvent(event beat.Event) bool { return true }
+func (o Outlet) Close() error                  { return nil }
+func (o Outlet) Done() <-chan struct{}         { return nil }
+
+// Connector is a connector to a test empty outlet.
+var Connector = channel.ConnectorFunc(
+	func(_ *common.Config, _ beat.ClientConfig) (channel.Outleter, error) {
+		return Outlet{}, nil
+	},
+)
+
+// AssertNotStartedInputCanBeDone checks that the context of an input can be
+// done before starting the input, and it doesn't leak goroutines. This is
+// important to confirm that leaks don't happen with CheckConfig.
+func AssertNotStartedInputCanBeDone(t *testing.T, factory input.Factory, configMap *common.MapStr) {
+	goroutines := resources.NewGoroutinesChecker()
+	defer goroutines.Check(t)
+
+	config, err := common.NewConfigFrom(configMap)
+	require.NoError(t, err)
+
+	context := input.Context{
+		Done: make(chan struct{}),
+	}
+
+	_, err = factory(config, Connector, context)
+	assert.NoError(t, err)
+
+	close(context.Done)
+}

filebeat/input/kafka/input_test.go

@@ -0,0 +1,36 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// +build !integration
+
+package kafka
+
+import (
+	"testing"
+
+	"github.com/elastic/beats/v7/filebeat/input/inputtest"
+	"github.com/elastic/beats/v7/libbeat/common"
+)
+
+func TestNewInputDone(t *testing.T) {
+	config := common.MapStr{
+		"hosts":    "localhost:9092",
+		"topics":   "messages",
+		"group_id": "filebeat",
+	}
+	inputtest.AssertNotStartedInputCanBeDone(t, NewInput, &config)
+}

filebeat/input/log/input_other_test.go

@@ -22,10 +22,11 @@ package log
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/elastic/beats/v7/filebeat/input/file"
+	"github.com/elastic/beats/v7/filebeat/input/inputtest"
 	"github.com/elastic/beats/v7/libbeat/common/match"
-
-	"github.com/stretchr/testify/assert"
 )
 
 var matchTests = []struct {
@@ -148,7 +149,7 @@ func TestInit(t *testing.T) {
 				Paths: test.paths,
 			},
 			states:              file.NewStates(),
-			outlet:              TestOutlet{},
+			outlet:              inputtest.Outlet{},
 			fileStateIdentifier: &file.MockIdentifier{},
 		}
 

filebeat/input/log/input_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/elastic/beats/v7/filebeat/channel"
 	"github.com/elastic/beats/v7/filebeat/input"
 	"github.com/elastic/beats/v7/filebeat/input/file"
+	"github.com/elastic/beats/v7/filebeat/input/inputtest"
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/common/match"
@@ -185,25 +186,10 @@ func testInputLifecycle(t *testing.T, context input.Context, closer func(input.C
 }
 
 func TestNewInputDone(t *testing.T) {
-	goroutines := resources.NewGoroutinesChecker()
-	defer goroutines.Check(t)
-
-	config, _ := common.NewConfigFrom(common.MapStr{
+	config := common.MapStr{
 		"paths": path.Join(os.TempDir(), "logs", "*.log"),
-	})
-
-	connector := channel.ConnectorFunc(func(_ *common.Config, _ beat.ClientConfig) (channel.Outleter, error) {
-		return TestOutlet{}, nil
-	})
-
-	context := input.Context{
-		Done: make(chan struct{}),
 	}
-
-	_, err := NewInput(config, connector, context)
-	assert.NoError(t, err)
-
-	close(context.Done)
+	inputtest.AssertNotStartedInputCanBeDone(t, NewInput, &config)
 }
 
 func TestNewInputError(t *testing.T) {
@@ -213,7 +199,7 @@ func TestNewInputError(t *testing.T) {
 	config := common.NewConfig()
 
 	connector := channel.ConnectorFunc(func(_ *common.Config, _ beat.ClientConfig) (channel.Outleter, error) {
-		return TestOutlet{}, nil
+		return inputtest.Outlet{}, nil
 	})
 
 	context := input.Context{}
@@ -318,10 +304,3 @@ func (o *eventCapturer) Close() error {
 func (o *eventCapturer) Done() <-chan struct{} {
 	return o.c
 }
-
-// TestOutlet is an empty outlet for testing
-type TestOutlet struct{}
-
-func (o TestOutlet) OnEvent(event beat.Event) bool { return true }
-func (o TestOutlet) Close() error                  { return nil }
-func (o TestOutlet) Done() <-chan struct{}         { return nil }

filebeat/input/mqtt/input_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	finput "github.com/elastic/beats/v7/filebeat/input"
+	"github.com/elastic/beats/v7/filebeat/input/inputtest"
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/common/backoff"
@@ -322,6 +323,13 @@ func TestOnCreateHandler_SubscribeMultiple_BackoffSignalDone(t *testing.T) {
 	require.Equal(t, 1, mockedBackoff.resetCount)
 }
 
+func TestNewInputDone(t *testing.T) {
+	config := common.MapStr{
+		"hosts": "tcp://:0",
+	}
+	inputtest.AssertNotStartedInputCanBeDone(t, NewInput, &config)
+}
+
 func assertEventMatches(t *testing.T, expected mockedMessage, got beat.Event) {
 	topic, err := got.GetValue("mqtt.topic")
 	require.NoError(t, err)

filebeat/input/redis/input_test.go

@@ -0,0 +1,34 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// +build !integration
+
+package redis
+
+import (
+	"testing"
+
+	"github.com/elastic/beats/v7/filebeat/input/inputtest"
+	"github.com/elastic/beats/v7/libbeat/common"
+)
+
+func TestNewInputDone(t *testing.T) {
+	config := common.MapStr{
+		"hosts": "localhost:3679",
+	}
+	inputtest.AssertNotStartedInputCanBeDone(t, NewInput, &config)
+}