[chore][stanza/namedpipe] Refactor the namedpipe input to remove the fsnotify (#39529)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

The current logic is that the input starts 2 goroutines, one for
watching for write events, and another one listens from `watcher.C` and
kick off the process. In other words, it relies on a write event to
start to read from the pipe.

When an active writer is writing to the pipe, and the collector exits
for whatever reason (or simply can't consume fast enough), the pipe
buffer can be filled and no write can come in. When no write happens,
the watcher doesn't receive any write event, and won't kick off the
process to read the pipe. This is effectively a deadlock, the writer
can't write to the pipe because the pipe buffer is full, and the
collector won't read from the pipe because no write event triggered.

Refactor the namedpipe input to get rid of the watcher. The input will
now keep reading from pipe. If no data or no external writer, it blocks
at the `scan.Scan()` wait for new data or an error when pipe is closed.

#### How to reproduce the bug
A simple python script to write to a named pipe periodically:

```python
import os
import time

PIPE_PATH = "/tmp/my_named_pipe"

try:
    fd = os.open(PIPE_PATH, os.O_RDWR | os.O_APPEND | os.O_NONBLOCK)
except OSError as e:
    print(f"Failed to open pipe: {e}")
    exit(1)

msg = b"x" * 1024 * 10 + b"\n" # 10 KB message with a newline at the end

while True:
    try:
        os.write(fd, msg)
        print("write ok")
    except BlockingIOError:
        print("PIPE FULL - write blocked")
    time.sleep(1)
```

The minimal Otel collector config:
```yaml
receivers:
  namedpipe:
    path: /tmp/my_named_pipe
    mode: 0600

exporters:
  debug:
    verbosity: basic

service:
  pipelines:
    logs:
      receivers: [namedpipe]
      processors: []
      exporters: [debug]
```

1. Start the python script to write the pipe to full (when seeing "PIPE
FULL - write blocked"). It's usually the 7th write if the pipe buffer is
64K (default value).
2. Start the otel collector and see it doesn't process anything from the
pipe.

<!--Describe what testing was performed and which tests were added.-->
#### Testing
Passed all current unit tests. Also tested locally with the steps above
to confirm it reads from the pipe from the start.

---------

Signed-off-by: Mengnan Gong <namco1992@gmail.com>
Co-authored-by: Daniel Jaglowski <jaglows3@gmail.com>

Author: namco1992

pkg/stanza/go.mod

@@ -6,7 +6,6 @@ require (
 	github.com/bmatcuk/doublestar/v4 v4.8.1
 	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/expr-lang/expr v1.17.2
-	github.com/fsnotify/fsnotify v1.9.0
 	github.com/goccy/go-json v0.10.5
 	github.com/jonboulle/clockwork v0.5.0
 	github.com/jpillora/backoff v1.0.0
@@ -45,6 +44,7 @@ require (
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/elastic/lunes v0.1.0 // indirect
+	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect

pkg/stanza/operator/input/namedpipe/input.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"os"
 	"sync"
+	"time"
 
 	"go.uber.org/zap"
 	"golang.org/x/sys/unix"
@@ -20,6 +21,8 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/trim"
 )
 
+const ReadTimeout = 2 * time.Second
+
 type Input struct {
 	helper.InputOperator
 
@@ -55,59 +58,61 @@ func (i *Input) Start(_ operator.Persister) error {
 		return fmt.Errorf("failed to chmod named pipe: %w", chmodErr)
 	}
 
-	watcher, err := NewWatcher(i.path)
-	if err != nil {
-		return fmt.Errorf("failed to create watcher: %w", err)
-	}
-
+	// Open the pipe with O_RDWR so it won't block on opening the pipe when there is no writer.
+	// The current process is both a writer and reader, which prevents the read from receiving
+	// EOF because there is always a writer (the process itself) connects to the pipe.
 	pipe, err := os.OpenFile(i.path, os.O_RDWR, os.ModeNamedPipe)
 	if err != nil {
 		return fmt.Errorf("failed to open named pipe: %w", err)
 	}
-
 	i.pipe = pipe
 
 	ctx, cancel := context.WithCancel(context.Background())
 	i.cancel = cancel
 
-	i.wg.Add(2)
-	go func() {
-		defer i.wg.Done()
-		if err := watcher.Watch(ctx); err != nil {
-			i.Logger().Error("failed to watch named pipe", zap.Error(err))
-		}
-	}()
-
-	go func() {
-		defer i.wg.Done()
-		for {
-			select {
-			case <-watcher.C:
-				if err := i.process(ctx, pipe); err != nil {
-					i.Logger().Error("failed to process named pipe", zap.Error(err))
-				}
-			case <-ctx.Done():
-				return
-			}
-		}
-	}()
+	i.wg.Add(1)
+	go i.readLoop(ctx)
 
 	return nil
 }
 
 func (i *Input) Stop() error {
-	if i.pipe != nil {
-		i.pipe.Close()
-	}
-
 	if i.cancel != nil {
 		i.cancel()
 	}
 
+	if i.pipe != nil {
+		i.pipe.Close()
+	}
+
 	i.wg.Wait()
 	return nil
 }
 
+func (i *Input) readLoop(ctx context.Context) {
+	defer i.wg.Done()
+	pipe := i.pipe
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			if err := i.process(ctx, pipe); err != nil {
+				i.Logger().Error("error processing named pipe", zap.Error(err))
+			}
+
+			// The process exits due to whatever reason, wait for ReadTimeout and try again.
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(ReadTimeout):
+				i.Logger().Warn("processing named pipe is interrupted, retrying the process now", zap.String("path", i.path))
+			}
+		}
+	}
+}
+
 func (i *Input) process(ctx context.Context, pipe *os.File) error {
 	scan := bufio.NewScanner(pipe)
 	scan.Split(i.splitFunc)

pkg/stanza/operator/input/namedpipe/input_test.go

@@ -150,3 +150,60 @@ func TestPipeWrites(t *testing.T) {
 		}
 	}
 }
+
+// TestPipeHasDataAtStartup will test if the receiver can consume from a named
+// pipe that has buffered data before startup.
+func TestPipeHasDataAtStartup(t *testing.T) {
+	fake := testutil.NewFakeOutput(t)
+
+	conf := NewConfig()
+	conf.Path = filename(t)
+	conf.Permissions = 0o666
+	conf.OutputIDs = []string{fake.ID()}
+
+	set := componenttest.NewNopTelemetrySettings()
+	op, err := conf.Build(set)
+	require.NoError(t, err)
+	ops := []operator.Operator{op, fake}
+
+	// create pipe
+	require.NoError(t, unix.Mkfifo(conf.Path, conf.Permissions))
+
+	pipe, err := os.OpenFile(conf.Path, os.O_RDWR|os.O_APPEND, os.ModeNamedPipe)
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, pipe.Close())
+	}()
+
+	logs := []string{"log1\n", "log2\n"}
+
+	for _, log := range logs {
+		_, err = pipe.WriteString(log)
+		require.NoError(t, err)
+	}
+
+	p, err := pipeline.NewDirectedPipeline(ops)
+	require.NoError(t, err)
+
+	// start receiver
+	require.NoError(t, p.Start(testutil.NewUnscopedMockPersister()))
+	defer func() {
+		require.NoError(t, p.Stop())
+	}()
+
+	for _, log := range logs {
+		expect := &entry.Entry{
+			Body: strings.TrimSpace(log),
+		}
+
+		select {
+		case e := <-fake.Received:
+			obs := time.Now()
+			expect.ObservedTimestamp = obs
+			e.ObservedTimestamp = obs
+			require.Equal(t, expect, e)
+		case <-time.After(time.Second):
+			t.Fatal("timed-out waiting for log entry")
+		}
+	}
+}

pkg/stanza/operator/input/namedpipe/watcher.go

@@ -31,7 +31,6 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/elastic/lunes v0.1.0 // indirect
 	github.com/expr-lang/expr v1.17.2 // indirect
-	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect