Add NamedPipe input operator

This is a namedpipe input operator, which will read from a named pipe
and send the data to the pipeline. It pretty closely mimicks the file input operator,
but with a few differences.

In particular, named pipes have an interesting property that they receive
EOFs when a writer closes the pipe, but that _doesn't_ mean that the pipe
is closed. To solve this issue, we crib from existing `tail -f` implementations
and use an inotify watcher to detect whenever the pipe receives new data, and
then read it using the standard `bufio.Scanner` reader.

Signed-off-by: sinkingpoint <colin@quirl.co.nz>

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

@@ -0,0 +1,196 @@
+package namedpipe
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"os"
+	"sync"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/decode"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/split"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/trim"
+	"go.uber.org/zap"
+	"golang.org/x/sys/unix"
+)
+
+const (
+	operatorType      = "namedpipe"
+	DefaultMaxLogSize = 1024 * 1024
+)
+
+func init() {
+	operator.Register(operatorType, func() operator.Builder { return NewConfig() })
+}
+
+func NewConfig() *Config {
+	return NewConfigWithID(operatorType)
+}
+
+// NewConfig creates a new stdin input config with default values
+func NewConfigWithID(operatorID string) *Config {
+	return &Config{
+		InputConfig: helper.NewInputConfig(operatorID, operatorType),
+	}
+}
+
+// Config is the configuration of a stdin input operator.
+type Config struct {
+	helper.InputConfig `mapstructure:",squash"`
+	BaseConfig         `mapstructure:",squash"`
+}
+
+type BaseConfig struct {
+	Path        string       `mapstructure:"path"`
+	Permissions uint32       `mapstructure:"mode"`
+	Encoding    string       `mapstructure:"encoding"`
+	SplitConfig split.Config `mapstructure:"multiline,omitempty"`
+	TrimConfig  trim.Config  `mapstructure:",squash"`
+}
+
+// Build will build a namedpipe input operator.
+func (c *Config) Build(logger *zap.SugaredLogger) (operator.Operator, error) {
+	inputOperator, err := c.InputConfig.Build(logger)
+	if err != nil {
+		return nil, err
+	}
+
+	enc, err := decode.LookupEncoding(c.Encoding)
+	if err != nil {
+		return nil, fmt.Errorf("failed to lookup encoding %q: %w", c.Encoding, err)
+	}
+
+	splitFunc, err := c.SplitConfig.Func(enc, true, DefaultMaxLogSize)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create split function: %w", err)
+	}
+
+	return &Input{
+		InputOperator: inputOperator,
+
+		buffer:      make([]byte, DefaultMaxLogSize),
+		path:        c.Path,
+		permissions: c.Permissions,
+		splitFunc:   splitFunc,
+		trimFunc:    c.TrimConfig.Func(),
+	}, nil
+}
+
+type Input struct {
+	helper.InputOperator
+
+	buffer      []byte
+	path        string
+	permissions uint32
+	splitFunc   bufio.SplitFunc
+	trimFunc    trim.Func
+	cancel      context.CancelFunc
+	pipe        *os.File
+	wg          sync.WaitGroup
+}
+
+func (n *Input) Start(p operator.Persister) error {
+	stat, err := os.Stat(n.path)
+	if err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("failed to stat named pipe: %w", err)
+	}
+
+	if !os.IsNotExist(err) && stat.Mode()&os.ModeNamedPipe == 0 {
+		return fmt.Errorf("path %s is not a named pipe", n.path)
+	}
+
+	if os.IsNotExist(err) {
+		if err := unix.Mkfifo(n.path, n.permissions); err != nil {
+			return fmt.Errorf("failed to create named pipe: %w", err)
+		}
+	}
+
+	// chmod the named pipe because mkfifo respects the umask which may result
+	// in a named pipe with incorrect permissions.
+	if err := os.Chmod(n.path, os.FileMode(n.permissions)); err != nil {
+		return fmt.Errorf("failed to chmod named pipe: %w", err)
+	}
+
+	watcher, err := NewWatcher(n.path)
+	if err != nil {
+		return fmt.Errorf("failed to create watcher: %w", err)
+	}
+
+	pipe, err := os.OpenFile(n.path, os.O_RDWR, os.ModeNamedPipe)
+	if err != nil {
+		return fmt.Errorf("failed to open named pipe: %w", err)
+	}
+
+	n.pipe = pipe
+
+	ctx, cancel := context.WithCancel(context.Background())
+	n.cancel = cancel
+
+	n.wg.Add(2)
+	go func() {
+		defer n.wg.Done()
+		if err := watcher.Watch(ctx); err != nil {
+			n.Logger().Errorw("failed to watch named pipe", zap.Error(err))
+		}
+	}()
+
+	go func() {
+		defer n.wg.Done()
+		for {
+			select {
+			case <-watcher.C:
+				if err := n.process(ctx, pipe); err != nil {
+					n.Logger().Errorw("failed to process named pipe", zap.Error(err))
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (n *Input) Stop() error {
+	n.pipe.Close()
+	n.cancel()
+	n.wg.Wait()
+	return nil
+}
+
+func (n *Input) process(ctx context.Context, pipe *os.File) error {
+	scan := bufio.NewScanner(pipe)
+	scan.Split(n.splitFunc)
+	scan.Buffer(n.buffer, len(n.buffer))
+
+	for scan.Scan() {
+		line := scan.Bytes()
+		if len(line) == 0 {
+			continue
+		}
+
+		if err := n.sendEntry(ctx, line); err != nil {
+			return fmt.Errorf("failed to send entry: %w", err)
+		}
+	}
+
+	return scan.Err()
+}
+
+// sendEntry sends an entry to the next operator in the pipeline.
+func (n *Input) sendEntry(ctx context.Context, bytes []byte) error {
+	bytes = n.trimFunc(bytes)
+	if len(bytes) == 0 {
+		return nil
+	}
+
+	entry, err := n.NewEntry(string(bytes))
+	if err != nil {
+		return fmt.Errorf("failed to create entry: %w", err)
+	}
+
+	n.Write(ctx, entry)
+	return nil
+}

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

@@ -0,0 +1,104 @@
+package namedpipe
+
+import (
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/pipeline"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zaptest"
+)
+
+// filename attempts to get an unused filename.
+func filename(t testing.TB) string {
+	t.Helper()
+
+	file, err := os.CreateTemp("", "")
+	require.NoError(t, err)
+
+	name := file.Name()
+	require.NoError(t, file.Close())
+	require.NoError(t, os.Remove(name))
+
+	return name
+}
+
+// TestCreatePipe tests that the named pipe is created as a pipe with the correct permissions.
+func TestCreatePipe(t *testing.T) {
+	conf := NewConfig()
+	conf.Path = filename(t)
+	conf.Permissions = 0666
+
+	op, err := conf.Build(zaptest.NewLogger(t).Sugar())
+	require.NoError(t, err)
+
+	require.NoError(t, op.Start(testutil.NewUnscopedMockPersister()))
+	defer func() {
+		require.NoError(t, op.Stop())
+	}()
+
+	stat, err := os.Stat(conf.Path)
+	require.NoError(t, err)
+
+	isPipe := stat.Mode()&os.ModeNamedPipe != 0
+	require.True(t, isPipe, "file is not a named pipe")
+	require.Equal(t, conf.Permissions, uint32(stat.Mode().Perm()))
+}
+
+// TestPipeWrites writes a few logs to the pipe over a few different connections and verifies that they are received.
+func TestPipeWrites(t *testing.T) {
+	fake := testutil.NewFakeOutput(t)
+
+	conf := NewConfig()
+	conf.Path = filename(t)
+	conf.Permissions = 0666
+	conf.OutputIDs = []string{fake.ID()}
+
+	op, err := conf.Build(zaptest.NewLogger(t).Sugar())
+	require.NoError(t, err)
+	ops := []operator.Operator{op, fake}
+
+	p, err := pipeline.NewDirectedPipeline(ops)
+	require.NoError(t, err)
+
+	require.NoError(t, p.Start(testutil.NewUnscopedMockPersister()))
+	defer p.Stop()
+
+	logs := [][]string{
+		{"log1\n", "log2\n"},
+		{"log3\n", "log4\n"},
+		{"log5\n"},
+	}
+
+	for _, toSend := range logs {
+		pipe, err := os.OpenFile(conf.Path, os.O_WRONLY, 0)
+		require.NoError(t, err)
+		defer pipe.Close()
+
+		for _, log := range toSend {
+			_, err = pipe.WriteString(log)
+			require.NoError(t, err)
+		}
+
+		for _, log := range toSend {
+			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 entry")
+			}
+		}
+	}
+}

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

@@ -0,0 +1,61 @@
+package namedpipe
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/fsnotify/fsnotify"
+)
+
+// Watcher watches a file for writes, notifying via `C` when one is observed.
+type Watcher struct {
+	C       chan struct{}
+	watcher *fsnotify.Watcher
+}
+
+// NewWatcher creates a new watcher for the given path.
+func NewWatcher(path string) (*Watcher, error) {
+	watcher, err := newWatcher(path)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Watcher{
+		C:       make(chan struct{}),
+		watcher: watcher,
+	}, nil
+}
+
+// Watch starts the watcher, sending a message to `C` when a write is observed.
+func (w *Watcher) Watch(ctx context.Context) error {
+	defer func() {
+		w.watcher.Close()
+		close(w.C)
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case event := <-w.watcher.Events:
+			if event.Op&fsnotify.Write == fsnotify.Write {
+				w.C <- struct{}{}
+			}
+		case err := <-w.watcher.Errors:
+			return fmt.Errorf("watcher error: %w", err)
+		}
+	}
+}
+
+func newWatcher(path string) (*fsnotify.Watcher, error) {
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create watcher: %w", err)
+	}
+
+	if err := watcher.Add(path); err != nil {
+		return nil, fmt.Errorf("failed to add path to watcher: %w", err)
+	}
+
+	return watcher, nil
+}