VAULT-17073: file sink node (#21817)

* audit file sink node

* Added options for file sink, updated tests,

* Ported benchmark for file

* tests

internal/observability/event/audit_sink_file.go

@@ -0,0 +1,249 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package event
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+
+	"github.com/hashicorp/eventlogger"
+)
+
+// defaultFileMode is the default file permissions (read/write for everyone).
+const (
+	defaultFileMode = 0o600
+	discard         = "discard"
+	stdout          = "stdout"
+)
+
+// AuditFileSink is a sink node which handles writing audit events to file.
+type AuditFileSink struct {
+	file     *os.File
+	fileLock sync.RWMutex
+	fileMode os.FileMode
+	path     string
+	format   auditFormat
+	prefix   string
+}
+
+// NewAuditFileSink should be used to create a new AuditFileSink.
+func NewAuditFileSink(path string, format auditFormat, opt ...Option) (*AuditFileSink, error) {
+	const op = "event.NewAuditFileSink"
+
+	// Parse and check path
+	p := strings.TrimSpace(path)
+	switch {
+	case p == "":
+		return nil, fmt.Errorf("%s: path is required", op)
+	case strings.EqualFold(path, stdout):
+		p = stdout
+	case strings.EqualFold(path, discard):
+		p = discard
+	}
+
+	// Validate format
+	if err := format.validate(); err != nil {
+		return nil, fmt.Errorf("%s: invalid format: %w", op, err)
+	}
+
+	opts, err := getOpts(opt...)
+	if err != nil {
+		return nil, fmt.Errorf("%s: error applying options: %w", op, err)
+	}
+
+	mode := os.FileMode(defaultFileMode)
+	// If we got an optional file mode supplied and our path isn't a special keyword
+	// then we should use the supplied file mode, or maintain the existing file mode.
+	if opts.withFileMode != nil {
+		switch {
+		case p == stdout:
+		case p == discard:
+		case *opts.withFileMode == 0: // Maintain the existing file's mode when set to "0000".
+			fileInfo, err := os.Stat(path)
+			if err != nil {
+				return nil, fmt.Errorf("%s: unable to determine existing file mode: %w", op, err)
+			}
+			mode = fileInfo.Mode()
+		default:
+			mode = *opts.withFileMode
+		}
+	}
+
+	return &AuditFileSink{
+		file:     nil,
+		fileLock: sync.RWMutex{},
+		fileMode: mode,
+		format:   format,
+		path:     p,
+		prefix:   opts.withPrefix,
+	}, nil
+}
+
+// Process handles writing the event to the file sink.
+func (f *AuditFileSink) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
+	const op = "event.(AuditFileSink).Process"
+
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	default:
+	}
+
+	if e == nil {
+		return nil, fmt.Errorf("%s: event is nil: %w", op, ErrInvalidParameter)
+	}
+
+	// 'discard' path means we just do nothing and pretend we're done.
+	if f.path == discard {
+		return nil, nil
+	}
+
+	formatted, found := e.Format(f.format.String())
+	if !found {
+		return nil, fmt.Errorf("%s: unable to retrieve event formatted as %q", op, f.format)
+	}
+
+	buffer := bytes.NewBuffer(formatted)
+	err := f.log(buffer)
+	if err != nil {
+		return nil, fmt.Errorf("%s: error writing file for audit sink: %w", op, err)
+	}
+
+	return nil, nil
+}
+
+// Reopen handles closing and reopening the file.
+func (f *AuditFileSink) Reopen() error {
+	const op = "event.(AuditFileSink).Reopen"
+
+	switch f.path {
+	case stdout, discard:
+		return nil
+	}
+
+	f.fileLock.Lock()
+	defer f.fileLock.Unlock()
+
+	if f.file == nil {
+		return f.open()
+	}
+
+	err := f.file.Close()
+	// Set to nil here so that even if we error out, on the next access open() will be tried.
+	f.file = nil
+	if err != nil {
+		return fmt.Errorf("%s: unable to close file for re-opening on audit sink: %w", op, err)
+	}
+
+	return f.open()
+}
+
+// Type is used to define which type of node AuditFileSink is.
+func (f *AuditFileSink) Type() eventlogger.NodeType {
+	return eventlogger.NodeTypeSink
+}
+
+// open attempts to open a file at the sink's path, with the sink's fileMode permissions
+// if one is not already open.
+// It doesn't have any locking and relies on calling functions of AuditFileSink to
+// handle this (e.g. log and Reopen methods).
+func (f *AuditFileSink) open() error {
+	const op = "event.(AuditFileSink).open"
+
+	if f.file != nil {
+		return nil
+	}
+
+	if err := os.MkdirAll(filepath.Dir(f.path), f.fileMode); err != nil {
+		return fmt.Errorf("%s: unable to create file %q: %w", op, f.path, err)
+	}
+
+	var err error
+	f.file, err = os.OpenFile(f.path, os.O_APPEND|os.O_WRONLY|os.O_CREATE, f.fileMode)
+	if err != nil {
+		return fmt.Errorf("%s: unable to open file for audit sink: %w", op, err)
+	}
+
+	// Change the file mode in case the log file already existed.
+	// We special case '/dev/null' since we can't chmod it, and bypass if the mode is zero.
+	switch f.path {
+	case "/dev/null":
+	default:
+		if f.fileMode != 0 {
+			err = os.Chmod(f.path, f.fileMode)
+			if err != nil {
+				return fmt.Errorf("%s: unable to change file %q permissions '%v' for audit sink: %w", op, f.path, f.fileMode, err)
+			}
+		}
+	}
+
+	return nil
+}
+
+// log writes the buffer to the file.
+// It acquires a lock on the file to do this.
+func (f *AuditFileSink) log(buf *bytes.Buffer) error {
+	const op = "event.(AuditFileSink).log"
+
+	f.fileLock.Lock()
+	defer f.fileLock.Unlock()
+
+	reader := bytes.NewReader(buf.Bytes())
+
+	var writer io.Writer
+	switch {
+	case f.path == stdout:
+		writer = os.Stdout
+	default:
+		if err := f.open(); err != nil {
+			return fmt.Errorf("%s: unable to open file for audit sink: %w", op, err)
+		}
+		writer = f.file
+	}
+
+	// Write prefix before the data if required.
+	if f.prefix != "" {
+		_, err := writer.Write([]byte(f.prefix))
+		if err != nil {
+			return fmt.Errorf("%s: unable to write prefix %q for audit sink: %w", op, f.prefix, err)
+		}
+	}
+
+	if _, err := reader.WriteTo(writer); err == nil {
+		return nil
+	} else if f.path == stdout {
+		// If writing to stdout there's no real reason to think anything would change on retry.
+		return fmt.Errorf("%s: unable write to %q: %w", op, f.path, err)
+	}
+
+	// Otherwise, opportunistically try to re-open the FD, once per call (1 retry attempt).
+	err := f.file.Close()
+	if err != nil {
+		return fmt.Errorf("%s: unable to close file for audit sink: %w", op, err)
+	}
+
+	f.file = nil
+
+	if err := f.open(); err != nil {
+		return fmt.Errorf("%s: unable to re-open file for audit sink: %w", op, err)
+	}
+
+	_, err = reader.Seek(0, io.SeekStart)
+	if err != nil {
+		return fmt.Errorf("%s: unable to seek to start of file for audit sink: %w", op, err)
+	}
+
+	_, err = reader.WriteTo(writer)
+	if err != nil {
+		return fmt.Errorf("%s: unable to re-write to file for audit sink: %w", op, err)
+	}
+
+	return nil
+}

internal/observability/event/audit_sink_file_test.go

@@ -0,0 +1,415 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package event
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/vault/sdk/logical"
+
+	vaultaudit "github.com/hashicorp/vault/audit"
+
+	"github.com/hashicorp/vault/helper/namespace"
+
+	"github.com/hashicorp/eventlogger"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestAuditFileSink_Type ensures that the node is a 'sink' type.
+func TestAuditFileSink_Type(t *testing.T) {
+	f, err := NewAuditFileSink(t.TempDir(), AuditFormatJSON)
+	require.NoError(t, err)
+	require.NotNil(t, f)
+	require.Equal(t, eventlogger.NodeTypeSink, f.Type())
+}
+
+// TestNewAuditFileSink tests creation of an AuditFileSink.
+func TestNewAuditFileSink(t *testing.T) {
+	tests := map[string]struct {
+		IsTempDirPath        bool // Path should contain the filename if temp dir is true
+		Path                 string
+		Format               auditFormat
+		Options              []Option
+		IsErrorExpected      bool
+		ExpectedErrorMessage string
+		// Expected values of AuditFileSink
+		ExpectedFileMode os.FileMode
+		ExpectedFormat   auditFormat
+		ExpectedPath     string
+		ExpectedPrefix   string
+	}{
+		"default-values": {
+			IsErrorExpected:      true,
+			ExpectedErrorMessage: "event.NewAuditFileSink: path is required",
+		},
+		"spacey-path": {
+			Path:                 "     ",
+			Format:               AuditFormatJSON,
+			IsErrorExpected:      true,
+			ExpectedErrorMessage: "event.NewAuditFileSink: path is required",
+		},
+		"bad-format": {
+			Path:                 "qwerty",
+			Format:               "squirrels",
+			IsErrorExpected:      true,
+			ExpectedErrorMessage: "event.NewAuditFileSink: invalid format: event.(auditFormat).validate: 'squirrels' is not a valid format: invalid parameter",
+		},
+		"path-not-exist-valid-format-file-mode": {
+			Path:             "qwerty",
+			Format:           AuditFormatJSON,
+			Options:          []Option{WithFileMode("00755")},
+			IsErrorExpected:  false,
+			ExpectedPath:     "qwerty",
+			ExpectedFormat:   AuditFormatJSON,
+			ExpectedPrefix:   "",
+			ExpectedFileMode: os.FileMode(0o755),
+		},
+		"valid-path-no-format": {
+			IsTempDirPath:        true,
+			Path:                 "vault.log",
+			IsErrorExpected:      true,
+			ExpectedErrorMessage: "event.NewAuditFileSink: invalid format: event.(auditFormat).validate: '' is not a valid format: invalid parameter",
+		},
+		"valid-path-and-format": {
+			IsTempDirPath:    true,
+			Path:             "vault.log",
+			Format:           AuditFormatJSON,
+			IsErrorExpected:  false,
+			ExpectedFileMode: defaultFileMode,
+			ExpectedFormat:   AuditFormatJSON,
+			ExpectedPrefix:   "",
+		},
+		"file-mode-not-default-or-zero": {
+			Path:             "vault.log",
+			Format:           AuditFormatJSON,
+			Options:          []Option{WithFileMode("0007")},
+			IsTempDirPath:    true,
+			IsErrorExpected:  false,
+			ExpectedFormat:   AuditFormatJSON,
+			ExpectedPrefix:   "",
+			ExpectedFileMode: 0o007,
+		},
+		"path-stdout": {
+			Path:             "stdout",
+			Format:           AuditFormatJSON,
+			Options:          []Option{WithFileMode("0007")}, // Will be ignored as stdout
+			IsTempDirPath:    false,
+			IsErrorExpected:  false,
+			ExpectedPath:     "stdout",
+			ExpectedFormat:   AuditFormatJSON,
+			ExpectedPrefix:   "",
+			ExpectedFileMode: defaultFileMode,
+		},
+		"path-discard": {
+			Path:             "discard",
+			Format:           AuditFormatJSON,
+			Options:          []Option{WithFileMode("0007")},
+			IsTempDirPath:    false,
+			IsErrorExpected:  false,
+			ExpectedPath:     "discard",
+			ExpectedFormat:   AuditFormatJSON,
+			ExpectedPrefix:   "",
+			ExpectedFileMode: defaultFileMode,
+		},
+		"prefix": {
+			IsTempDirPath:    true,
+			Path:             "vault.log",
+			Format:           AuditFormatJSON,
+			Options:          []Option{WithFileMode("0007"), WithPrefix("bleep")},
+			IsErrorExpected:  false,
+			ExpectedPrefix:   "bleep",
+			ExpectedFormat:   AuditFormatJSON,
+			ExpectedFileMode: 0o007,
+		},
+	}
+
+	for name, tc := range tests {
+		name := name
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			// t.Parallel()
+
+			// If we need a real directory as a path we can use a temp dir.
+			// but we should keep track of it for comparison in the new sink.
+			var tempDir string
+			tempPath := tc.Path
+			if tc.IsTempDirPath {
+				tempDir = t.TempDir()
+				tempPath = filepath.Join(tempDir, tempPath)
+			}
+
+			sink, err := NewAuditFileSink(tempPath, tc.Format, tc.Options...)
+
+			switch {
+			case tc.IsErrorExpected:
+				require.Error(t, err)
+				require.EqualError(t, err, tc.ExpectedErrorMessage)
+				require.Nil(t, sink)
+			default:
+				require.NoError(t, err)
+				require.NotNil(t, sink)
+
+				// Assert properties are correct.
+				require.Equal(t, tc.ExpectedPrefix, sink.prefix)
+				require.Equal(t, tc.ExpectedFormat, sink.format)
+				require.Equal(t, tc.ExpectedFileMode, sink.fileMode)
+
+				switch {
+				case tc.IsTempDirPath:
+					require.Equal(t, tempPath, sink.path)
+				default:
+					require.Equal(t, tc.ExpectedPath, sink.path)
+				}
+			}
+		})
+	}
+}
+
+// TestAuditFileSink_Reopen tests that the sink reopens files as expected when requested to.
+// stdout and discard paths are ignored.
+// see: https://developer.hashicorp.com/vault/docs/audit/file#file_path
+func TestAuditFileSink_Reopen(t *testing.T) {
+	tests := map[string]struct {
+		Path                 string
+		IsTempDirPath        bool
+		ShouldCreateFile     bool
+		Options              []Option
+		IsErrorExpected      bool
+		ExpectedErrorMessage string
+		ExpectedFileMode     os.FileMode
+	}{
+		// Should be ignored by Reopen
+		"discard": {
+			Path: "discard",
+		},
+		// Should be ignored by Reopen
+		"stdout": {
+			Path: "stdout",
+		},
+		"permission-denied": {
+			Path:                 "/tmp/vault/test/foo.log",
+			IsErrorExpected:      true,
+			ExpectedErrorMessage: "event.(AuditFileSink).open: unable to create file \"/tmp/vault/test/foo.log\": mkdir /tmp/vault/test: permission denied",
+		},
+		"happy": {
+			Path:             "vault.log",
+			IsTempDirPath:    true,
+			ExpectedFileMode: os.FileMode(defaultFileMode),
+		},
+		"filemode-existing": {
+			Path:             "vault.log",
+			IsTempDirPath:    true,
+			ShouldCreateFile: true,
+			Options:          []Option{WithFileMode("0000")},
+			ExpectedFileMode: os.FileMode(defaultFileMode),
+		},
+	}
+
+	for name, tc := range tests {
+		name := name
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			// If we need a real directory as a path we can use a temp dir.
+			// but we should keep track of it for comparison in the new sink.
+			var tempDir string
+			tempPath := tc.Path
+			if tc.IsTempDirPath {
+				tempDir = t.TempDir()
+				tempPath = filepath.Join(tempDir, tc.Path)
+			}
+
+			// If the file mode is 0 then we will need a pre-created file to stat.
+			// Only do this for paths that are not 'special keywords'
+			if tc.ShouldCreateFile && tc.Path != discard && tc.Path != stdout {
+				f, err := os.OpenFile(tempPath, os.O_CREATE, defaultFileMode)
+				require.NoError(t, err)
+				defer func() {
+					err = os.Remove(f.Name())
+					require.NoError(t, err)
+				}()
+			}
+
+			sink, err := NewAuditFileSink(tempPath, AuditFormatJSON, tc.Options...)
+			require.NoError(t, err)
+			require.NotNil(t, sink)
+
+			err = sink.Reopen()
+
+			switch {
+			case tc.IsErrorExpected:
+				require.Error(t, err)
+				require.EqualError(t, err, tc.ExpectedErrorMessage)
+			case tempPath == discard:
+				require.NoError(t, err)
+			case tempPath == stdout:
+				require.NoError(t, err)
+			default:
+				require.NoError(t, err)
+				info, err := os.Stat(tempPath)
+				require.NoError(t, err)
+				require.NotNil(t, info)
+				require.Equal(t, tc.ExpectedFileMode, info.Mode())
+			}
+		})
+	}
+}
+
+// TestAuditFileSink_Process ensures that Process behaves as expected.
+func TestAuditFileSink_Process(t *testing.T) {
+	tests := map[string]struct {
+		Path                 string
+		Format               auditFormat
+		Data                 *logical.LogInput
+		ShouldIgnoreFormat   bool
+		ShouldUseNilEvent    bool
+		IsErrorExpected      bool
+		ExpectedErrorMessage string
+	}{
+		"discard": {
+			Path:            discard,
+			Format:          AuditFormatJSON,
+			Data:            &logical.LogInput{Request: &logical.Request{ID: "123"}},
+			IsErrorExpected: false,
+		},
+		"stdout": {
+			Path:            stdout,
+			Format:          AuditFormatJSON,
+			Data:            &logical.LogInput{Request: &logical.Request{ID: "123"}},
+			IsErrorExpected: false,
+		},
+		"no-formatted-data": {
+			Path:                 "/dev/null",
+			Format:               AuditFormatJSON,
+			Data:                 &logical.LogInput{Request: &logical.Request{ID: "123"}},
+			ShouldIgnoreFormat:   true,
+			IsErrorExpected:      true,
+			ExpectedErrorMessage: "event.(AuditFileSink).Process: unable to retrieve event formatted as \"json\"",
+		},
+		"nil": {
+			Path:                 "/dev/null",
+			Format:               AuditFormatJSON,
+			Data:                 &logical.LogInput{Request: &logical.Request{ID: "123"}},
+			ShouldUseNilEvent:    true,
+			IsErrorExpected:      true,
+			ExpectedErrorMessage: "event.(AuditFileSink).Process: event is nil: invalid parameter",
+		},
+	}
+
+	for name, tc := range tests {
+		name := name
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			// Setup a formatter
+			cfg := vaultaudit.FormatterConfig{}
+			ss := newStaticSalt(t)
+			formatter, err := NewAuditFormatterJSON(cfg, ss)
+			require.NoError(t, err)
+			require.NotNil(t, formatter)
+
+			// Setup a sink
+			sink, err := NewAuditFileSink(tc.Path, tc.Format)
+			require.NoError(t, err)
+			require.NotNil(t, sink)
+
+			// Generate a fake event
+			ctx := namespace.RootContext(nil)
+			event := fakeJSONAuditEvent(t, AuditRequest, tc.Data)
+			require.NotNil(t, event)
+
+			// Finesse the event into the correct shape.
+			event, err = formatter.Process(ctx, event)
+			require.NoError(t, err)
+			require.NotNil(t, event)
+
+			// Some conditional stuff 'per test' to exercise different parts of Process.
+			if tc.ShouldIgnoreFormat {
+				delete(event.Formatted, tc.Format.String())
+			}
+
+			if tc.ShouldUseNilEvent {
+				event = nil
+			}
+
+			// The actual exercising of the sink.
+			event, err = sink.Process(ctx, event)
+			switch {
+			case tc.IsErrorExpected:
+				require.Error(t, err)
+				require.EqualError(t, err, tc.ExpectedErrorMessage)
+				require.Nil(t, event)
+			default:
+				require.NoError(t, err)
+				require.Nil(t, event)
+			}
+		})
+	}
+}
+
+// BenchmarkAuditFileSink_Process benchmarks the AuditFormatterJSON and then AuditFileSink calling Process.
+// This should replicate the original benchmark testing which used to perform both of these roles together.
+func BenchmarkAuditFileSink_Process(b *testing.B) {
+	// Base input
+	in := &logical.LogInput{
+		Auth: &logical.Auth{
+			ClientToken:     "foo",
+			Accessor:        "bar",
+			EntityID:        "foobarentity",
+			DisplayName:     "testtoken",
+			NoDefaultPolicy: true,
+			Policies:        []string{"root"},
+			TokenType:       logical.TokenTypeService,
+		},
+		Request: &logical.Request{
+			Operation: logical.UpdateOperation,
+			Path:      "/foo",
+			Connection: &logical.Connection{
+				RemoteAddr: "127.0.0.1",
+			},
+			WrapInfo: &logical.RequestWrapInfo{
+				TTL: 60 * time.Second,
+			},
+			Headers: map[string][]string{
+				"foo": {"bar"},
+			},
+		},
+	}
+
+	ctx := namespace.RootContext(nil)
+
+	// Create the formatter node.
+	cfg := vaultaudit.FormatterConfig{}
+	ss := newStaticSalt(b)
+	formatter, err := NewAuditFormatterJSON(cfg, ss)
+	require.NoError(b, err)
+	require.NotNil(b, formatter)
+
+	// Create the sink node.
+	sink, err := NewAuditFileSink("/dev/null", AuditFormatJSON)
+	require.NoError(b, err)
+	require.NotNil(b, sink)
+
+	// Generate the event
+	event := fakeJSONAuditEvent(b, AuditRequest, in)
+	require.NotNil(b, event)
+
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			event, err = formatter.Process(ctx, event)
+			if err != nil {
+				panic(err)
+			}
+			_, err := sink.Process(ctx, event)
+			if err != nil {
+				panic(err)
+			}
+		}
+	})
+}

internal/observability/event/event_type_audit.go

@@ -64,9 +64,9 @@ func newAudit(opt ...Option) (*audit, error) {
 	audit := &audit{
 		ID:             opts.withID,
 		Version:        auditVersion,
-		Subtype:        auditSubtype(opts.withSubtype),
+		Subtype:        opts.withSubtype,
 		Timestamp:      opts.withNow,
-		RequiredFormat: auditFormat(opts.withFormat),
+		RequiredFormat: opts.withFormat,
 	}
 
 	if err := audit.validate(); err != nil {
@@ -109,7 +109,7 @@ func (a *audit) validate() error {
 
 // validate ensures that auditSubtype is one of the set of allowed event subtypes.
 func (t auditSubtype) validate() error {
-	const op = "event.(audit).(subtype).validate"
+	const op = "event.(auditSubtype).validate"
 	switch t {
 	case AuditRequest, AuditResponse:
 		return nil
@@ -120,12 +120,12 @@ func (t auditSubtype) validate() error {
 
 // validate ensures that auditFormat is one of the set of allowed event formats.
 func (f auditFormat) validate() error {
-	const op = "event.(audit).(format).validate"
+	const op = "event.(auditFormat).validate"
 	switch f {
 	case AuditFormatJSON, AuditFormatJSONx:
 		return nil
 	default:
-		return fmt.Errorf("%s: '%s' is not a valid required format: %w", op, f, ErrInvalidParameter)
+		return fmt.Errorf("%s: '%s' is not a valid format: %w", op, f, ErrInvalidParameter)
 	}
 }
 

internal/observability/event/event_type_audit_test.go

@@ -25,12 +25,12 @@ func TestAuditEvent_New(t *testing.T) {
 		"nil": {
 			Options:              nil,
 			IsErrorExpected:      true,
-			ExpectedErrorMessage: "event.newAudit: event.(audit).validate: event.(audit).(subtype).validate: '' is not a valid event subtype: invalid parameter",
+			ExpectedErrorMessage: "event.newAudit: event.(audit).validate: event.(auditSubtype).validate: '' is not a valid event subtype: invalid parameter",
 		},
 		"empty-option": {
 			Options:              []Option{},
 			IsErrorExpected:      true,
-			ExpectedErrorMessage: "event.newAudit: event.(audit).validate: event.(audit).(subtype).validate: '' is not a valid event subtype: invalid parameter",
+			ExpectedErrorMessage: "event.newAudit: event.(audit).validate: event.(auditSubtype).validate: '' is not a valid event subtype: invalid parameter",
 		},
 		"bad-id": {
 			Options:              []Option{WithID("")},
@@ -145,7 +145,7 @@ func TestAuditEvent_Validate(t *testing.T) {
 				RequiredFormat: AuditFormatJSON,
 			},
 			IsErrorExpected:      true,
-			ExpectedErrorMessage: "event.(audit).validate: event.(audit).(subtype).validate: 'moon' is not a valid event subtype: invalid parameter",
+			ExpectedErrorMessage: "event.(audit).validate: event.(auditSubtype).validate: 'moon' is not a valid event subtype: invalid parameter",
 		},
 		"format-fiddled": {
 			Value: &audit{
@@ -157,7 +157,7 @@ func TestAuditEvent_Validate(t *testing.T) {
 				RequiredFormat: auditFormat("blah"),
 			},
 			IsErrorExpected:      true,
-			ExpectedErrorMessage: "event.(audit).validate: event.(audit).(format).validate: 'blah' is not a valid required format: invalid parameter",
+			ExpectedErrorMessage: "event.(audit).validate: event.(auditFormat).validate: 'blah' is not a valid format: invalid parameter",
 		},
 		"default-time": {
 			Value: &audit{
@@ -212,12 +212,12 @@ func TestAuditEvent_Validate_Subtype(t *testing.T) {
 		"empty": {
 			Value:                "",
 			IsErrorExpected:      true,
-			ExpectedErrorMessage: "event.(audit).(subtype).validate: '' is not a valid event subtype: invalid parameter",
+			ExpectedErrorMessage: "event.(auditSubtype).validate: '' is not a valid event subtype: invalid parameter",
 		},
 		"unsupported": {
 			Value:                "foo",
 			IsErrorExpected:      true,
-			ExpectedErrorMessage: "event.(audit).(subtype).validate: 'foo' is not a valid event subtype: invalid parameter",
+			ExpectedErrorMessage: "event.(auditSubtype).validate: 'foo' is not a valid event subtype: invalid parameter",
 		},
 		"request": {
 			Value:           "AuditRequest",
@@ -257,12 +257,12 @@ func TestAuditEvent_Validate_Format(t *testing.T) {
 		"empty": {
 			Value:                "",
 			IsErrorExpected:      true,
-			ExpectedErrorMessage: "event.(audit).(format).validate: '' is not a valid required format: invalid parameter",
+			ExpectedErrorMessage: "event.(auditFormat).validate: '' is not a valid format: invalid parameter",
 		},
 		"unsupported": {
 			Value:                "foo",
 			IsErrorExpected:      true,
-			ExpectedErrorMessage: "event.(audit).(format).validate: 'foo' is not a valid required format: invalid parameter",
+			ExpectedErrorMessage: "event.(auditFormat).validate: 'foo' is not a valid format: invalid parameter",
 		},
 		"json": {
 			Value:           "json",

internal/observability/event/node_formatter_audit_json.go

@@ -54,13 +54,14 @@ func (_ *AuditFormatterJSON) Type() eventlogger.NodeType {
 // Process will attempt to parse the incoming event data into a corresponding
 // audit request/response entry which is serialized to JSON and stored within the event.
 func (f *AuditFormatterJSON) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
+	const op = "event.(AuditFormatterJSON).Process"
+
 	select {
 	case <-ctx.Done():
 		return nil, ctx.Err()
 	default:
 	}
 
-	const op = "event.(AuditFormatterJSON).Process"
 	if e == nil {
 		return nil, fmt.Errorf("%s: event is nil: %w", op, ErrInvalidParameter)
 	}

internal/observability/event/node_formatter_audit_json_test.go

@@ -22,8 +22,8 @@ import (
 
 // fakeJSONAuditEvent will return a new fake event containing audit data based
 // on the specified auditSubtype and logical.LogInput.
-func fakeJSONAuditEvent(t *testing.T, subtype auditSubtype, input *logical.LogInput) *eventlogger.Event {
+func fakeJSONAuditEvent(tb testing.TB, subtype auditSubtype, input *logical.LogInput) *eventlogger.Event {
-	t.Helper()
+	tb.Helper()
 
 	date := time.Date(2023, time.July, 11, 15, 49, 10, 0o0, time.Local)
 
@@ -33,13 +33,13 @@ func fakeJSONAuditEvent(t *testing.T, subtype auditSubtype, input *logical.LogIn
 		WithFormat(string(AuditFormatJSON)),
 		WithNow(date),
 	)
-	require.NoError(t, err)
+	require.NoError(tb, err)
-	require.NotNil(t, auditEvent)
+	require.NotNil(tb, auditEvent)
-	require.Equal(t, "123", auditEvent.ID)
+	require.Equal(tb, "123", auditEvent.ID)
-	require.Equal(t, "v0.1", auditEvent.Version)
+	require.Equal(tb, "v0.1", auditEvent.Version)
-	require.Equal(t, AuditFormatJSON, auditEvent.RequiredFormat)
+	require.Equal(tb, AuditFormatJSON, auditEvent.RequiredFormat)
-	require.Equal(t, subtype, auditEvent.Subtype)
+	require.Equal(tb, subtype, auditEvent.Subtype)
-	require.Equal(t, date, auditEvent.Timestamp)
+	require.Equal(tb, date, auditEvent.Timestamp)
 
 	auditEvent.Data = input
 
@@ -54,9 +54,9 @@ func fakeJSONAuditEvent(t *testing.T, subtype auditSubtype, input *logical.LogIn
 }
 
 // newStaticSalt returns a new staticSalt for use in testing.
-func newStaticSalt(t *testing.T) *staticSalt {
+func newStaticSalt(tb testing.TB) *staticSalt {
 	s, err := salt.NewSalt(context.Background(), nil, nil)
-	require.NoError(t, err)
+	require.NoError(tb, err)
 
 	return &staticSalt{salt: s}
 }

internal/observability/event/node_formatter_audit_jsonx.go

@@ -40,13 +40,14 @@ func (_ *AuditFormatterJSONx) Type() eventlogger.NodeType {
 // Process will attempt to retrieve pre-formatted JSON stored within the event
 // and re-encode the data to JSONx.
 func (f *AuditFormatterJSONx) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
+	const op = "event.(AuditFormatterJSONx).Process"
+
 	select {
 	case <-ctx.Done():
 		return nil, ctx.Err()
 	default:
 	}
 
-	const op = "event.(AuditFormatterJSONx).Process"
 	if e == nil {
 		return nil, fmt.Errorf("%s: event is nil: %w", op, ErrInvalidParameter)
 	}

internal/observability/event/options.go

@@ -6,6 +6,8 @@ package event
 import (
 	"errors"
 	"fmt"
+	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -19,8 +21,10 @@ type Option func(*options) error
 type options struct {
 	withID       string
 	withNow      time.Time
-	withSubtype string
+	withSubtype  auditSubtype
-	withFormat  string
+	withFormat   auditFormat
+	withFileMode *os.FileMode
+	withPrefix   string
 }
 
 // getDefaultOptions returns options with their default values.
@@ -63,7 +67,7 @@ func NewID(prefix string) (string, error) {
 	return fmt.Sprintf("%s_%s", prefix, id), nil
 }
 
-// WithID allows an optional ID.
+// WithID provides an optional ID.
 func WithID(id string) Option {
 	return func(o *options) error {
 		var err error
@@ -80,7 +84,7 @@ func WithID(id string) Option {
 	}
 }
 
-// WithNow allows an option to represent 'now'.
+// WithNow provides an option to represent 'now'.
 func WithNow(now time.Time) Option {
 	return func(o *options) error {
 		var err error
@@ -96,36 +100,78 @@ func WithNow(now time.Time) Option {
 	}
 }
 
-// WithSubtype allows an option to represent the subtype.
+// WithSubtype provides an option to represent the subtype.
 func WithSubtype(subtype string) Option {
 	return func(o *options) error {
-		var err error
+		s := strings.TrimSpace(subtype)
-
+		if s == "" {
-		subtype := strings.TrimSpace(subtype)
+			return errors.New("subtype cannot be empty")
-		switch {
-		case subtype == "":
-			err = errors.New("subtype cannot be empty")
-		default:
-			o.withSubtype = subtype
 		}
 
+		parsed := auditSubtype(s)
+		err := parsed.validate()
+		if err != nil {
 			return err
 		}
+
+		o.withSubtype = parsed
+		return nil
+	}
 }
 
-// WithFormat allows an option to represent event format.
+// WithFormat provides an option to represent event format.
 func WithFormat(format string) Option {
 	return func(o *options) error {
-		var err error
+		f := strings.TrimSpace(format)
-
+		if f == "" {
-		format := strings.TrimSpace(format)
+			return errors.New("format cannot be empty")
-		switch {
-		case format == "":
-			err = errors.New("format cannot be empty")
-		default:
-			o.withFormat = format
 		}
 
+		parsed := auditFormat(f)
+		err := parsed.validate()
+		if err != nil {
 			return err
 		}
+
+		o.withFormat = parsed
+		return nil
+	}
+}
+
+// WithFileMode provides an option to represent a file mode for a file sink.
+// Supplying an empty string or whitespace will prevent this option from being
+// applied, but it will not return an error in those circumstances.
+func WithFileMode(mode string) Option {
+	return func(o *options) error {
+		// Clear up whitespace before attempting to parse
+		mode = strings.TrimSpace(mode)
+		if mode == "" {
+			// If supplied file mode is empty, just return early without setting anything.
+			// We can assume that this option was called by something that didn't
+			// parse the incoming value, perhaps from a config map etc.
+			return nil
+		}
+
+		// By now we believe we have something that the caller really intended to
+		// be parsed into a file mode.
+		raw, err := strconv.ParseUint(mode, 8, 32)
+
+		switch {
+		case err != nil:
+			return fmt.Errorf("unable to parse file mode: %w", err)
+		default:
+			m := os.FileMode(raw)
+			o.withFileMode = &m
+		}
+
+		return nil
+	}
+}
+
+// WithPrefix provides an option to represent a prefix for a file sink.
+func WithPrefix(prefix string) Option {
+	return func(o *options) error {
+		o.withPrefix = prefix
+		return nil
+	}
 }

internal/observability/event/options_test.go

@@ -4,6 +4,7 @@
 package event
 
 import (
+	"os"
 	"testing"
 	"time"
 
@@ -16,7 +17,7 @@ func TestOptions_WithFormat(t *testing.T) {
 		Value                string
 		IsErrorExpected      bool
 		ExpectedErrorMessage string
-		ExpectedValue        string
+		ExpectedValue        auditFormat
 	}{
 		"empty": {
 			Value:                "",
@@ -28,10 +29,20 @@ func TestOptions_WithFormat(t *testing.T) {
 			IsErrorExpected:      true,
 			ExpectedErrorMessage: "format cannot be empty",
 		},
-		"valid": {
+		"invalid-test": {
 			Value:                "test",
+			IsErrorExpected:      true,
+			ExpectedErrorMessage: "event.(auditFormat).validate: 'test' is not a valid format: invalid parameter",
+		},
+		"valid-json": {
+			Value:           "json",
 			IsErrorExpected: false,
-			ExpectedValue:   "test",
+			ExpectedValue:   AuditFormatJSON,
+		},
+		"valid-jsonx": {
+			Value:           "jsonx",
+			IsErrorExpected: false,
+			ExpectedValue:   AuditFormatJSONx,
 		},
 	}
 
@@ -61,7 +72,7 @@ func TestOptions_WithSubtype(t *testing.T) {
 		Value                string
 		IsErrorExpected      bool
 		ExpectedErrorMessage string
-		ExpectedValue        string
+		ExpectedValue        auditSubtype
 	}{
 		"empty": {
 			Value:                "",
@@ -74,9 +85,9 @@ func TestOptions_WithSubtype(t *testing.T) {
 			ExpectedErrorMessage: "subtype cannot be empty",
 		},
 		"valid": {
-			Value:           "test",
+			Value:           "AuditResponse",
 			IsErrorExpected: false,
-			ExpectedValue:   "test",
+			ExpectedValue:   AuditResponse,
 		},
 	}
 
@@ -186,6 +197,69 @@ func TestOptions_WithID(t *testing.T) {
 	}
 }
 
+// TestOptions_WithFileMode exercises WithFileMode option to ensure it performs as expected.
+func TestOptions_WithFileMode(t *testing.T) {
+	tests := map[string]struct {
+		Value                string
+		IsErrorExpected      bool
+		ExpectedErrorMessage string
+		IsNilExpected        bool
+		ExpectedValue        os.FileMode
+	}{
+		"empty": {
+			Value:           "",
+			IsErrorExpected: false,
+			IsNilExpected:   true,
+		},
+		"whitespace": {
+			Value:           "     ",
+			IsErrorExpected: false,
+			IsNilExpected:   true,
+		},
+		"nonsense": {
+			Value:                "juan",
+			IsErrorExpected:      true,
+			ExpectedErrorMessage: "unable to parse file mode: strconv.ParseUint: parsing \"juan\": invalid syntax",
+		},
+		"zero": {
+			Value:           "0000",
+			IsErrorExpected: false,
+			ExpectedValue:   os.FileMode(0o000),
+		},
+		"valid": {
+			Value:           "0007",
+			IsErrorExpected: false,
+			ExpectedValue:   os.FileMode(0o007),
+		},
+	}
+
+	for name, tc := range tests {
+		name := name
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			options := &options{}
+			applyOption := WithFileMode(tc.Value)
+			err := applyOption(options)
+			switch {
+			case tc.IsErrorExpected:
+				require.Error(t, err)
+				require.EqualError(t, err, tc.ExpectedErrorMessage)
+			default:
+				require.NoError(t, err)
+				switch {
+				case tc.IsNilExpected:
+					// Optional option 'not supplied' (i.e. was whitespace/empty string)
+					require.Nil(t, options.withFileMode)
+				default:
+					// Dereference the pointer, so we can examine the file mode.
+					require.Equal(t, tc.ExpectedValue, *options.withFileMode)
+				}
+			}
+		})
+	}
+}
+
 // TestOptions_Default exercises getDefaultOptions to assert the default values.
 func TestOptions_Default(t *testing.T) {
 	opts := getDefaultOptions()
@@ -201,8 +275,8 @@ func TestOptions_Opts(t *testing.T) {
 		IsErrorExpected      bool
 		ExpectedErrorMessage string
 		ExpectedID           string
-		ExpectedSubtype      string
+		ExpectedSubtype      auditSubtype
-		ExpectedFormat       string
+		ExpectedFormat       auditFormat
 		IsNowExpected        bool
 		ExpectedNow          time.Time
 	}{
@@ -227,20 +301,20 @@ func TestOptions_Opts(t *testing.T) {
 		},
 		"with-multiple-valid-subtype": {
 			opts: []Option{
-				WithSubtype("qwerty"),
+				WithSubtype("AuditRequest"),
-				WithSubtype("juan"),
+				WithSubtype("AuditResponse"),
 			},
 			IsErrorExpected: false,
-			ExpectedSubtype: "juan",
+			ExpectedSubtype: AuditResponse,
 			IsNowExpected:   true,
 		},
 		"with-multiple-valid-format": {
 			opts: []Option{
-				WithFormat("qwerty"),
+				WithFormat("json"),
-				WithFormat("juan"),
+				WithFormat("jsonx"),
 			},
 			IsErrorExpected: false,
-			ExpectedFormat:  "juan",
+			ExpectedFormat:  AuditFormatJSONx,
 			IsNowExpected:   true,
 		},
 		"with-multiple-valid-now": {
@@ -263,14 +337,14 @@ func TestOptions_Opts(t *testing.T) {
 		"with-multiple-valid-options": {
 			opts: []Option{
 				WithID("qwerty"),
-				WithSubtype("typey2"),
+				WithSubtype("AuditRequest"),
 				WithFormat("json"),
 				WithNow(time.Date(2023, time.July, 4, 12, 3, 0, 0, time.Local)),
 			},
 			IsErrorExpected: false,
 			ExpectedID:      "qwerty",
-			ExpectedSubtype: "typey2",
+			ExpectedSubtype: AuditRequest,
-			ExpectedFormat:  "json",
+			ExpectedFormat:  AuditFormatJSON,
 			ExpectedNow:     time.Date(2023, time.July, 4, 12, 3, 0, 0, time.Local),
 		},
 	}