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, and
a regression test asserts Stream() leaves the channel open after teardown.

Author: alxrxs

pkg/acquisition/modules/syslog/run.go

@@ -43,8 +43,11 @@ func (s *Source) Stream(ctx context.Context, out chan pipeline.Event) error {
 	})
 
 	g.Go(func() error {
-		defer close(out)
-
+		// out is the shared pipeline output channel, fed by every configured
+		// datasource; it is owned by the acquisition orchestrator, not by this
+		// Stream(), so it must NOT be closed here. Doing so races with any other
+		// datasource still sending on it (e.g. an in-flight appsec out-of-band
+		// request) and panics with "send on closed channel".
 		for {
 			select {
 			case <-ctx.Done():

pkg/acquisition/modules/syslog/syslog_test.go

@@ -167,3 +167,45 @@ disable_rfc_parser: true`,
 		})
 	}
 }
+
+// TestStreamDoesNotCloseOutput is a regression test: Stream() must never close
+// the output channel. That channel is shared — the acquisition orchestrator
+// hands the same channel to every configured datasource and owns its lifecycle
+// (shutdown is driven by context cancellation, not by closing it). If syslog
+// closes it, any other datasource still sending on it panics with "send on
+// closed channel". We assert the channel stays open once Stream is torn down by
+// a context cancellation.
+func TestStreamDoesNotCloseOutput(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	s := Source{}
+	config := `source: syslog
+listen_port: 4243
+listen_addr: 127.0.0.1`
+
+	err := s.Configure(ctx, []byte(config), log.WithField("type", ModuleName), metrics.AcquisitionMetricsLevelNone)
+	require.NoError(t, err)
+
+	out := make(chan pipeline.Event)
+
+	go func() {
+		_ = s.Stream(ctx, out)
+	}()
+
+	// let the syslog server come up, then tear Stream down
+	time.Sleep(500 * time.Millisecond)
+	cancel()
+
+	// A closed channel yields (zero, false) on receive; an open, idle one hits the
+	// default branch. Assert we never observe a close over the window. Receiving
+	// (rather than sending) can't itself panic, so this is a safe probe.
+	require.Never(t, func() bool {
+		select {
+		case _, ok := <-out:
+			return !ok
+		default:
+			return false
+		}
+	}, time.Second, 50*time.Millisecond, "Stream() must not close the shared output channel")
+}

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
 }