diff --git a/internal/cmd/commands/dev/dev.go b/internal/cmd/commands/dev/dev.go index 098964a19..4a2322cf4 100644 --- a/internal/cmd/commands/dev/dev.go +++ b/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") diff --git a/internal/cmd/commands/dev/flags_test.go b/internal/cmd/commands/dev/flags_test.go index 5345063c9..49b9b3c0f 100644 --- a/internal/cmd/commands/dev/flags_test.go +++ b/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() diff --git a/internal/cmd/config/config.go b/internal/cmd/config/config.go index 91b977746..a7b114a91 100644 --- a/internal/cmd/config/config.go +++ b/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. diff --git a/internal/cmd/config/config_test.go b/internal/cmd/config/config_test.go index 8e5b61afc..bff37d476 100644 --- a/internal/cmd/config/config_test.go +++ b/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() diff --git a/internal/daemon/cluster/handlers/testing.go b/internal/daemon/cluster/handlers/testing.go index 94a3e2e5b..f92ff5b1a 100644 --- a/internal/daemon/cluster/handlers/testing.go +++ b/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) diff --git a/internal/daemon/worker/testing.go b/internal/daemon/worker/testing.go index cc4fe4814..32d5688cd 100644 --- a/internal/daemon/worker/testing.go +++ b/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) diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index 8e2a1967a..b37735c14 100644 --- a/internal/daemon/worker/worker.go +++ b/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 { diff --git a/internal/errors/code.go b/internal/errors/code.go index e93204ce6..6b3011e53 100644 --- a/internal/errors/code.go +++ b/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. diff --git a/internal/errors/code_test.go b/internal/errors/code_test.go index c4b2cbc55..e2d40471f 100644 --- a/internal/errors/code_test.go +++ b/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) { diff --git a/internal/errors/info.go b/internal/errors/info.go index d932748af..e24b6d0cf 100644 --- a/internal/errors/info.go +++ b/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, + }, } diff --git a/internal/storage/doc.go b/internal/storage/doc.go new file mode 100644 index 000000000..8a01c37a1 --- /dev/null +++ b/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 diff --git a/internal/storage/options.go b/internal/storage/options.go new file mode 100644 index 000000000..e9dd1988f --- /dev/null +++ b/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 + } +} diff --git a/internal/storage/options_test.go b/internal/storage/options_test.go new file mode 100644 index 000000000..7864ed3bf --- /dev/null +++ b/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) + }) +} diff --git a/internal/storage/storage.go b/internal/storage/storage.go new file mode 100644 index 000000000..565f4f22c --- /dev/null +++ b/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 +}