diff --git a/pkg/acquisition/modules/no_shared_channel_close_test.go b/pkg/acquisition/modules/no_shared_channel_close_test.go new file mode 100644 index 00000000000..938a40162ff --- /dev/null +++ b/pkg/acquisition/modules/no_shared_channel_close_test.go @@ -0,0 +1,100 @@ +package modules + +import ( + "go/ast" + "go/parser" + "go/token" + "os" + "path/filepath" + "runtime" + "testing" + + "github.com/stretchr/testify/require" +) + +// TestDatasourcesDoNotCloseSharedOutputChannel statically checks every +// datasource module for `close(out)` inside a Stream() or StreamingAcquisition() +// method. `out` is the pipeline output channel shared across every configured +// datasource (see acquisition.StartAcquisition); it is owned by the acquisition +// orchestrator, not by any individual datasource, so a datasource closing it +// races with any other datasource still sending on it and panics with "send on +// closed channel" (this happened in production with the syslog datasource). +// +// This is independent of any one datasource and covers every module directory +// under pkg/acquisition/modules automatically, including future ones, without +// needing to configure or run any of them. +func TestDatasourcesDoNotCloseSharedOutputChannel(t *testing.T) { + _, thisFile, _, ok := runtime.Caller(0) + require.True(t, ok) + + modulesDir := filepath.Dir(thisFile) + + entries, err := os.ReadDir(modulesDir) + require.NoError(t, err) + + for _, entry := range entries { + if !entry.IsDir() { + continue + } + + moduleDir := filepath.Join(modulesDir, entry.Name()) + + moduleFiles, err := os.ReadDir(moduleDir) + require.NoError(t, err, "reading module %s", entry.Name()) + + fset := token.NewFileSet() + + for _, moduleFile := range moduleFiles { + if moduleFile.IsDir() || filepath.Ext(moduleFile.Name()) != ".go" { + continue + } + + path := filepath.Join(moduleDir, moduleFile.Name()) + + file, err := parser.ParseFile(fset, path, nil, 0) + require.NoError(t, err, "parsing %s", path) + + checkFileDoesNotCloseOut(t, fset, entry.Name(), path, file) + } + } +} + +// checkFileDoesNotCloseOut walks a single file's AST for Stream/StreamingAcquisition +// method declarations and fails if their body contains close(out). +func checkFileDoesNotCloseOut(t *testing.T, fset *token.FileSet, module string, path string, file *ast.File) { + t.Helper() + + for _, decl := range file.Decls { + fn, ok := decl.(*ast.FuncDecl) + if !ok || fn.Body == nil { + continue + } + + if fn.Name.Name != "Stream" && fn.Name.Name != "StreamingAcquisition" { + continue + } + + ast.Inspect(fn.Body, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + + ident, ok := call.Fun.(*ast.Ident) + if !ok || ident.Name != "close" || len(call.Args) != 1 { + return true + } + + arg, ok := call.Args[0].(*ast.Ident) + if ok && arg.Name == "out" { + t.Errorf( + "%s: %s.%s must not close(out) -- it is the shared acquisition "+ + "output channel, owned by the orchestrator, not by this datasource (%s:%d)", + module, module, fn.Name.Name, path, fset.Position(call.Pos()).Line, + ) + } + + return true + }) + } +} diff --git a/pkg/acquisition/modules/syslog/run.go b/pkg/acquisition/modules/syslog/run.go index e3851d6e1b7..1a678c3b451 100644 --- a/pkg/acquisition/modules/syslog/run.go +++ b/pkg/acquisition/modules/syslog/run.go @@ -43,8 +43,6 @@ func (s *Source) Stream(ctx context.Context, out chan pipeline.Event) error { }) g.Go(func() error { - defer close(out) - for { select { case <-ctx.Done(): diff --git a/pkg/acquisition/types/types.go b/pkg/acquisition/types/types.go index 776592826b0..050e23a577e 100644 --- a/pkg/acquisition/types/types.go +++ b/pkg/acquisition/types/types.go @@ -76,6 +76,10 @@ type Fetcher interface { // failures, but treat them as errors. The caller is responsible for supervising // Stream(), and restarting it as needed. There is currently no way to differentiate // retryable vs permanent errors. +// - never close the output channel: it is shared and fed by every configured +// datasource, so it is owned by the acquisition orchestrator, not by any single +// Stream() implementation. Closing it here would race with (and can panic) any +// other datasource still sending on it. type RestartableStreamer interface { // Start live acquisition (eg, tail a file) Stream(ctx context.Context, out chan pipeline.Event) error @@ -84,6 +88,9 @@ type RestartableStreamer interface { // Tailer has the same pupose as RestartableStreamer (provide ongoing events) but // is responsible for spawning its own goroutines, and handling errors and retries. // New datasources are expected to implement RestartableStreamer instead. +// +// As with RestartableStreamer, the output channel is shared and orchestrator-owned: +// an implementation must never close it. type Tailer interface { StreamingAcquisition(ctx context.Context, out chan pipeline.Event, acquisTomb *tomb.Tomb) error }