fix(syslog): don't close the shared acquisition output channel

The syslog datasource's Stream() does `defer close(out)` on the pipeline
output channel. That channel is shared: StartAcquisition() passes the same
`out` to every configured datasource, and it is owned by the acquisition
orchestrator, which never closes it (shutdown is driven by context
cancellation, then drainChan). syslog is the only datasource that closes it.

When syslog and another datasource run together, syslog's close races with
the other datasource still sending. This is reproducible with the AppSec
datasource: on shutdown/reload, an in-flight out-of-band request in
appsec_runner's handleOutBandInterrupt does `r.outChan <- evt` after syslog
has closed the channel, panicking with "send on closed channel" (seen in
production on 1.7.8 under a syslog+appsec config).

Remove the close. No consumer relies on it: runParse ranges via a
`select { case <-ctx.Done(); case ev := <-input }` loop and drainChan is a
non-blocking best-effort drain, so both terminate on context cancellation
regardless. This aligns syslog with every other datasource (file, journalctl,
docker, ...), none of which close the shared channel. The "never close out"
contract is documented on the RestartableStreamer and Tailer interfaces.

Adds TestDatasourcesDoNotCloseSharedOutputChannel: a static check (parses the
AST of every pkg/acquisition/modules/<name> directory, no per-module config or
runtime needed) asserting no Stream/StreamingAcquisition implementation calls
close(out). It covers every present and future datasource automatically and
fails naming the exact file/line if the bug recurs anywhere.

Author: alxrxs

pkg/acquisition/modules/no_shared_channel_close_test.go

@@ -0,0 +1,95 @@
+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())
+
+		fset := token.NewFileSet()
+
+		pkgs, err := parser.ParseDir(fset, moduleDir, func(fi os.FileInfo) bool {
+			return filepath.Ext(fi.Name()) == ".go"
+		}, 0)
+		require.NoError(t, err, "parsing module %s", entry.Name())
+
+		for _, pkg := range pkgs {
+			for path, file := range pkg.Files {
+				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
+		})
+	}
+}

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():

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
 }