Fix leak caused by input runners created when checking their configuration

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("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("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)
+}