feat(storage): Add interface, options and errors

internal/cmd/commands/dev/dev.go

@@ -105,6 +105,7 @@ type Command struct {
 	flagWorkerAuthWorkerRotationInterval time.Duration
 	flagWorkerAuthCaCertificateLifetime  time.Duration
 	flagWorkerAuthDebuggingEnabled       bool
+	flagWorkerRecordingStorageDir        string
 }
 
 func (c *Command) Synopsis() string {
@@ -368,6 +369,13 @@ func (c *Command) Flags() *base.FlagSets {
 		Target: &c.flagWorkerAuthStorageDir,
 		Usage:  "Specifies the directory to store worker authentication credentials in dev mode. Setting this will make use of file storage at the specified location; otherwise in-memory storage or a temporary directory will be used.",
 	})
+
+	f.StringVar(&base.StringVar{
+		Name:   "worker-recording-storage-dir",
+		Target: &c.flagWorkerRecordingStorageDir,
+		Usage:  "Specifies the directory to store worker session recordings in dev mode.",
+	})
+
 	f.BoolVar(&base.BoolVar{
 		Name:   "worker-auth-storage-skip-cleanup",
 		Target: &c.flagWorkerAuthStorageSkipCleanup,
@@ -455,6 +463,7 @@ func (c *Command) Run(args []string) int {
 	c.DevTargetDefaultPort = c.flagTargetDefaultPort
 	c.Config.Plugins.ExecutionDir = c.flagPluginExecutionDir
 	c.Config.Worker.AuthStoragePath = c.flagWorkerAuthStorageDir
+	c.Config.Worker.RecordingStoragePath = c.flagWorkerRecordingStorageDir
 	if c.flagIdSuffix != "" {
 		if len(c.flagIdSuffix) != 10 {
 			c.UI.Error("Invalid ID suffix, must be exactly 10 characters")

internal/cmd/commands/dev/flags_test.go

@@ -69,6 +69,7 @@ func TestCommand_Flags(t *testing.T) {
 	assert.Contains(completions, "-plugin-execution-dir")
 	assert.Contains(completions, "-worker-auth-method")
 	assert.Contains(completions, "-worker-auth-storage-dir")
+	assert.Contains(completions, "-worker-recording-storage-dir")
 	assert.Contains(completions, "-worker-auth-storage-skip-cleanup")
 
 	// keep adding assertions for other flags which should be set as a result of cmd.Flags()

internal/cmd/config/config.go

@@ -264,6 +264,10 @@ type Worker struct {
 	// AuthStoragePath represents the location a worker stores its node credentials, if set
 	AuthStoragePath string `hcl:"auth_storage_path"`
 
+	// RecordingStoragePath represents the location a worker caches session recordings before
+	// they are sync'ed to the corresponding storage bucket. The path must already exist.
+	RecordingStoragePath string `hcl:"recording_storage_path"`
+
 	// ControllerGeneratedActivationToken is a controller-generated activation
 	// token used to register this worker to the cluster. It can be a path, env
 	// var, or direct value.

internal/cmd/config/config_test.go

@@ -494,7 +494,7 @@ func TestDevCombined(t *testing.T) {
 	assert.Equal(t, exp, actual)
 }
 
-func TestDevWorkerCredentialStorageDir(t *testing.T) {
+func TestDevWorkerCredentialStoragePath(t *testing.T) {
 	t.Parallel()
 	tests := []struct {
 		name                           string
@@ -549,6 +549,62 @@ func TestDevWorkerCredentialStorageDir(t *testing.T) {
 	}
 }
 
+func TestDevWorkerRecordingStoragePath(t *testing.T) {
+	t.Parallel()
+	td := t.TempDir()
+	tests := []struct {
+		name                           string
+		devWorkerProvidedConfiguration string
+		storagePath                    string
+	}{
+		{
+			name: "Relative Storage Directory",
+			devWorkerProvidedConfiguration: `
+			listener "tcp" {
+				purpose = "proxy"
+			}
+
+			worker {
+				name = "w_1234567890"
+				description = "A default worker created in dev mode"
+				initial_upstreams = ["127.0.0.1"]
+				tags {
+					type = ["dev", "local"]
+				}
+				recording_storage_path = ".."
+			}
+			`,
+			storagePath: "..",
+		},
+		{
+			name: "temp dir",
+			devWorkerProvidedConfiguration: fmt.Sprintf(`
+			listener "tcp" {
+				purpose = "proxy"
+			}
+
+			worker {
+				name = "w_1234567890"
+				description = "A default worker created in dev mode"
+				initial_upstreams = ["127.0.0.1"]
+				tags {
+					type = ["dev", "local"]
+				}
+				recording_storage_path = "%v"
+			}
+			`, td),
+			storagePath: td,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			parsed, err := Parse(devConfig + tt.devWorkerProvidedConfiguration)
+			require.NoError(t, err)
+			require.Equal(t, tt.storagePath, parsed.Worker.RecordingStoragePath)
+		})
+	}
+}
+
 func TestDevKeyGeneration(t *testing.T) {
 	t.Parallel()
 	dk := DevKeyGeneration()

internal/daemon/cluster/handlers/testing.go

@@ -35,7 +35,7 @@ func TestUpstreamService(t *testing.T) (UpstreamMessageServiceClientProducer, *t
 	// sending upstream requests
 	initStorage, err := nodeefile.New(testCtx)
 	require.NoError(t, err)
-	t.Cleanup(initStorage.Cleanup)
+	t.Cleanup(func() { initStorage.Cleanup(testCtx) })
 	_, err = rotation.RotateRootCertificates(testCtx, initStorage)
 	require.NoError(t, err)
 	initNodeCreds, err := types.NewNodeCredentials(testCtx, initStorage)

internal/daemon/worker/testing.go

@@ -199,6 +199,9 @@ type TestWorkerOpts struct {
 	// The location of the worker's auth storage
 	WorkerAuthStoragePath string
 
+	// The location of the worker's recording storage
+	WorkerRecordingStoragePath string
+
 	// The name to use for the worker, otherwise one will be randomly
 	// generated, unless provided in a non-nil Config
 	Name string
@@ -305,6 +308,9 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker {
 	if opts.WorkerAuthStoragePath != "" {
 		opts.Config.Worker.AuthStoragePath = opts.WorkerAuthStoragePath
 	}
+	if opts.WorkerRecordingStoragePath != "" {
+		opts.Config.Worker.RecordingStoragePath = opts.WorkerRecordingStoragePath
+	}
 	tw.name = opts.Config.Worker.Name
 
 	if opts.SuccessfulStatusGracePeriodDuration != 0 {
@@ -500,6 +506,7 @@ func NewTestMultihopWorkers(t testing.TB,
 		InitialUpstreams:           kmsWorker.ProxyAddrs(),
 		Logger:                     logger.Named("childPkiWorker"),
 		Config:                     childPkiWorkerConf,
+		WorkerRecordingStoragePath: t.TempDir(),
 		WorkerAuthDebuggingEnabled: enableAuthDebugging,
 	})
 	t.Cleanup(childPkiWorker.Shutdown)

internal/daemon/worker/worker.go

@@ -29,6 +29,7 @@ import (
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/server"
+	"github.com/hashicorp/boundary/internal/storage"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-secure-stdlib/base62"
 	"github.com/hashicorp/go-secure-stdlib/mlock"
@@ -77,6 +78,8 @@ type recorderCache any
 // create its reverseConnReceiver
 var reverseConnReceiverFactory func() reverseConnReceiver
 
+var recordingStorageFactory func(ctx context.Context, path string) (storage.RecordingStorage, error)
+
 var recorderCacheFactory func() recorderCache
 
 var initializeReverseGrpcClientCollectors = noopInitializePromCollectors
@@ -138,6 +141,9 @@ type Worker struct {
 	WorkerAuthRegistrationRequest string
 	workerAuthSplitListener       *nodeenet.SplitListener
 
+	// The storage for session recording
+	RecordingStorage storage.RecordingStorage
+
 	// downstream workers and routes to those workers
 	downstreamWorkers  downstreamers
 	downstreamReceiver reverseConnReceiver
@@ -203,6 +209,14 @@ func New(conf *Config) (*Worker, error) {
 		conf.RawConfig.Worker = new(config.Worker)
 	}
 
+	if w.conf.RawConfig.Worker.RecordingStoragePath != "" && recordingStorageFactory != nil {
+		s, err := recordingStorageFactory(w.baseContext, w.conf.RawConfig.Worker.RecordingStoragePath)
+		if err != nil {
+			return nil, fmt.Errorf("error create recording storage: %w", err)
+		}
+		w.RecordingStorage = s
+	}
+
 	w.parseAndStoreTags(conf.RawConfig.Worker.Tags)
 
 	if conf.SecureRandomReader == nil {

internal/errors/code.go

@@ -48,11 +48,21 @@ const (
 	CycleFound                         = 121 // CycleFound represents an error when a cycle is found between a parent and child worker
 	WorkerConnNotFound                 = 122 // WorkerConnNotFound represents an error when a connection to a worker is not found
 	KmsWorkerUnsupportedOperation      = 123 // KmsWorkerUnsupportedOperation represents an error when a KMS worker is not supported for an operation
+
 	// Note: Currently unused in OSS
 	RetryLimitExceeded = 124 // RetryLimitExceeded represents an error when a retry limit is exceeded
 	// Note: Currently unused in OSS
 	QueueIsFull = 125 // QueueIsFull results in attempting to add an item to a queue which is full
 
+	// Note: Storage errors are currently unused in OSS
+	StorageFileClosed         = 126 // StorageFileClose represents an error when a file has been closed and a read/write operation is attempted on it
+	StorageContainerClosed    = 127 // StorageContainerClosed represents an error when a container has been closed and a read/write operation is attempted on it
+	StorageFileReadOnly       = 128 // StorageFileReadOnly represents an error when a file is readonly and a write operation is attempted on it
+	StorageFileWriteOnly      = 129 // StorageFileWriteOnly represents an error when a file is write only and a read operation is attempted on it
+	StorageFileAlreadyExists  = 130 // StorageFileAlreadyExists represents an error when a file already exists during an attempt to create it
+	StorageContainerReadOnly  = 131 // StorageContainerReadOnly represents an error when a container is readonly and a write operation is attempted on it
+	StorageContainerWriteOnly = 132 // StorageContainerWriteOnly represents an error when a container is write only and a read operation is attempted on it
+
 	AuthAttemptExpired Code = 198 // AuthAttemptExpired represents an expired authentication attempt
 	AuthMethodInactive Code = 199 // AuthMethodInactive represents an error that means the auth method is not active.
 

internal/errors/code_test.go

@@ -355,6 +355,41 @@ func TestCode_Both_String_Info(t *testing.T) {
 			c:    NotFound,
 			want: NotFound,
 		},
+		{
+			name: "StorageFileClose",
+			c:    StorageFileClosed,
+			want: StorageFileClosed,
+		},
+		{
+			name: "StorageContainerClosed",
+			c:    StorageContainerClosed,
+			want: StorageContainerClosed,
+		},
+		{
+			name: "StorageFileReadOnly",
+			c:    StorageFileReadOnly,
+			want: StorageFileReadOnly,
+		},
+		{
+			name: "StorageFileWriteOnly",
+			c:    StorageFileWriteOnly,
+			want: StorageFileWriteOnly,
+		},
+		{
+			name: "StorageFileAlreadyExists",
+			c:    StorageFileAlreadyExists,
+			want: StorageFileAlreadyExists,
+		},
+		{
+			name: "StorageContainerReadOnly",
+			c:    StorageContainerReadOnly,
+			want: StorageContainerReadOnly,
+		},
+		{
+			name: "StorageContainerWriteOnly",
+			c:    StorageContainerWriteOnly,
+			want: StorageContainerWriteOnly,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

internal/errors/info.go

@@ -299,4 +299,32 @@ var errorCodeInfo = map[Code]Info{
 		Message: "not found",
 		Kind:    State,
 	},
+	StorageFileClosed: {
+		Message: "file is closed",
+		Kind:    State,
+	},
+	StorageContainerClosed: {
+		Message: "container is closed",
+		Kind:    State,
+	},
+	StorageFileReadOnly: {
+		Message: "file is read only",
+		Kind:    State,
+	},
+	StorageFileWriteOnly: {
+		Message: "file is write only",
+		Kind:    State,
+	},
+	StorageFileAlreadyExists: {
+		Message: "file already exists",
+		Kind:    State,
+	},
+	StorageContainerReadOnly: {
+		Message: "container is read only",
+		Kind:    State,
+	},
+	StorageContainerWriteOnly: {
+		Message: "container is write only",
+		Kind:    State,
+	},
 }

internal/storage/doc.go

@@ -0,0 +1,6 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+// Package storage can be used for writing and reading files from storage buckets
+// within external object stores.
+package storage

internal/storage/options.go

@@ -0,0 +1,86 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package storage
+
+// SyncMode is used to determine how a file is synced when closed.
+type SyncMode uint8
+
+const (
+	// Asynchronous mode will trigger a file to sync to the storage
+	// bucket on a recurring interval once it has been closed.
+	// Asynchronous is the default sync mode.
+	Asynchronous SyncMode = iota
+
+	// Synchronous mode will cause a file closed call to block until the file
+	// has been synced to the storage bucket. Any error while syncing the file
+	// will be returned to the caller of Close.
+	Synchronous
+
+	// NoSync mode will result in Close not syncing the file to the storage
+	// bucket.
+	NoSync
+)
+
+// AccessMode is use to determine the access mode a file is opened with.
+type AccessMode uint8
+
+const (
+	// ReadOnly mode will open a file as read only.
+	// ReadOnly is the default access mode.
+	ReadOnly AccessMode = iota
+
+	// WriteOnly mode will open a file for writing only.
+	WriteOnly
+
+	// ReadWrite mode will open a file for reading and writing.
+	ReadWrite
+)
+
+// getOpts - iterate the inbound Options and return a struct.
+func getOpts(opt ...Option) Options {
+	opts := getDefaultOptions()
+	for _, o := range opt {
+		o(&opts)
+	}
+	return opts
+}
+
+func getDefaultOptions() Options {
+	return Options{
+		withCloseSyncMode:  Asynchronous,
+		withFileAccessMode: ReadOnly,
+		withCreateFile:     false,
+	}
+}
+
+// Options are storage options.
+type Options struct {
+	withCloseSyncMode  SyncMode
+	withFileAccessMode AccessMode
+	withCreateFile     bool
+}
+
+// Option is a storage option.
+type Option func(*Options)
+
+// WithCloseSyncMode sets how a file is synced when closed.
+func WithCloseSyncMode(m SyncMode) Option {
+	return func(o *Options) {
+		o.withCloseSyncMode = m
+	}
+}
+
+// WithCreateFile indicates a file should be created when opening.
+func WithCreateFile() Option {
+	return func(o *Options) {
+		o.withCreateFile = true
+	}
+}
+
+// WithFileAccessMode sets the access mode when a file is opened.
+func WithFileAccessMode(m AccessMode) Option {
+	return func(o *Options) {
+		o.withFileAccessMode = m
+	}
+}

internal/storage/options_test.go

@@ -0,0 +1,57 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package storage
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// Test_getOpts provides unit tests for GetOpts and all the options
+func Test_getOpts(t *testing.T) {
+	t.Parallel()
+	assert := assert.New(t)
+	t.Run("WithCloseSyncMode", func(t *testing.T) {
+		testOpts := getDefaultOptions()
+		opts := getOpts(WithCloseSyncMode(Asynchronous))
+		assert.Equal(testOpts, opts)
+
+		testOpts = getDefaultOptions()
+		opts = getOpts(WithCloseSyncMode(Synchronous))
+		testOpts.withCloseSyncMode = Synchronous
+		assert.Equal(opts, testOpts)
+
+		testOpts = getDefaultOptions()
+		opts = getOpts(WithCloseSyncMode(NoSync))
+		testOpts.withCloseSyncMode = NoSync
+		assert.Equal(opts, testOpts)
+	})
+	t.Run("WithFileAccessMode", func(t *testing.T) {
+		testOpts := getDefaultOptions()
+		opts := getOpts(WithFileAccessMode(ReadOnly))
+		assert.Equal(testOpts, opts)
+
+		testOpts = getDefaultOptions()
+		opts = getOpts(WithFileAccessMode(WriteOnly))
+		testOpts.withFileAccessMode = WriteOnly
+		assert.Equal(opts, testOpts)
+
+		testOpts = getDefaultOptions()
+		opts = getOpts(WithFileAccessMode(ReadWrite))
+		testOpts.withFileAccessMode = ReadWrite
+		assert.Equal(opts, testOpts)
+	})
+	t.Run("WithCreateFile", func(t *testing.T) {
+		testOpts := getDefaultOptions()
+		opts := getOpts()
+		testOpts.withCreateFile = false
+		assert.Equal(testOpts, opts)
+
+		testOpts = getDefaultOptions()
+		opts = getOpts(WithCreateFile())
+		testOpts.withCreateFile = true
+		assert.Equal(opts, testOpts)
+	})
+}

internal/storage/storage.go

@@ -0,0 +1,41 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package storage
+
+import (
+	"context"
+	"io"
+	"io/fs"
+)
+
+// RecordingStorage can be used to create a LocalFS usable for session recording.
+type RecordingStorage interface {
+	NewLocalFS(ctx context.Context, bucket Bucket, _ ...Option) (FS, error)
+}
+
+// Bucket is a resource that represents a bucket in an external object store
+type Bucket interface {
+	GetBucketName() string
+	GetBucketPrefix() string
+}
+
+// FS is a filesystem for creating or reading files and containers.
+type FS interface {
+	New(ctx context.Context, name string) (Container, error)
+}
+
+// A Container is a filesystem abstraction that can create files or other containers.
+type Container interface {
+	io.Closer
+	Create(context.Context, string) (File, error)
+	OpenFile(context.Context, string, ...Option) (File, error)
+	SubContainer(context.Context, string, ...Option) (Container, error)
+}
+
+// File represents a storage File.
+type File interface {
+	fs.File
+	io.Writer
+	io.StringWriter
+}