feat(storage): Add support for storage plugins

internal/cmd/base/base.go

@@ -35,16 +35,16 @@ type EnabledPlugin uint
 
 const (
 	EnabledPluginUnknown EnabledPlugin = iota
-	EnabledPluginHostLoopback
-	EnabledPluginHostAws
+	EnabledPluginLoopback
+	EnabledPluginAws
 	EnabledPluginHostAzure
 )
 
 func (e EnabledPlugin) String() string {
 	switch e {
-	case EnabledPluginHostLoopback:
+	case EnabledPluginLoopback:
 		return "Loopback"
-	case EnabledPluginHostAws:
+	case EnabledPluginAws:
 		return "AWS"
 	case EnabledPluginHostAzure:
 		return "Azure"

internal/cmd/commands/database/migrate.go

@@ -15,7 +15,7 @@ import (
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	host_plugin_assets "github.com/hashicorp/boundary/plugins/host"
-	external_host_plugins "github.com/hashicorp/boundary/sdk/plugins/host"
+	external_plugins "github.com/hashicorp/boundary/sdk/plugins"
 	"github.com/hashicorp/go-secure-stdlib/mlock"
 	"github.com/hashicorp/go-secure-stdlib/parseutil"
 	"github.com/hashicorp/go-secure-stdlib/pluginutil/v2"
@@ -179,14 +179,14 @@ func (c *MigrateCommand) Run(args []string) (retCode int) {
 		return base.CommandCliError
 	}
 
-	_, awsCleanup, err := external_host_plugins.CreateHostPlugin(
+	_, awsCleanup, err := external_plugins.CreateHostPlugin(
 		c.Context,
 		"aws",
-		external_host_plugins.WithPluginOptions(
+		external_plugins.WithPluginOptions(
 			pluginutil.WithPluginExecutionDirectory(c.Config.Plugins.ExecutionDir),
 			pluginutil.WithPluginsFilesystem(host_plugin_assets.HostPluginPrefix, host_plugin_assets.FileSystem()),
 		),
-		external_host_plugins.WithLogger(pluginLogger.Named("aws")),
+		external_plugins.WithLogger(pluginLogger.Named("aws")),
 	)
 	if err != nil {
 		c.UI.Error(fmt.Errorf("Error creating dynamic host plugin: %w", err).Error())

internal/cmd/commands/dev/dev.go

@@ -97,7 +97,7 @@ type Command struct {
 	flagSysEvents                        string
 	flagEveryEventAllowFilters           []string
 	flagEveryEventDenyFilters            []string
-	flagCreateLoopbackHostPlugin         bool
+	flagCreateLoopbackPlugin             bool
 	flagPluginExecutionDir               string
 	flagWorkerAuthMethod                 string
 	flagWorkerAuthStorageDir             string
@@ -381,7 +381,7 @@ func (c *Command) Flags() *base.FlagSets {
 	f.StringVar(&base.StringVar{
 		Name:   "worker-recording-storage-dir",
 		Target: &c.flagWorkerRecordingStorageDir,
-		Usage:  "Specifies the directory to store worker session recordings in dev mode.",
+		Usage:  "Specifies the directory to store worker session recordings in dev mode. If not provided a temp directory will be created. Session recording is an Enterprise-only feature.",
 	})
 
 	f.BoolVar(&base.BoolVar{
@@ -396,8 +396,8 @@ func (c *Command) Flags() *base.FlagSets {
 	})
 
 	f.BoolVar(&base.BoolVar{
-		Name:   "create-loopback-host-plugin",
-		Target: &c.flagCreateLoopbackHostPlugin,
+		Name:   "create-loopback-plugin",
+		Target: &c.flagCreateLoopbackPlugin,
 		Hidden: true,
 	})
 
@@ -480,7 +480,19 @@ 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.Config.Worker.RecordingStoragePath == "" {
+		// Create a temp dir for recording storage
+		const pattern = "recordingstorage"
+		c.Config.Worker.RecordingStoragePath, err = os.MkdirTemp("", pattern)
+		if err != nil {
+			c.UI.Error(fmt.Errorf("Error creating storage temp dir: %w", err).Error())
+			return base.CommandCliError
+		}
+		c.ShutdownFuncs = append(c.ShutdownFuncs, func() error { return os.RemoveAll(c.Config.Worker.RecordingStoragePath) })
+	}
+
 	if c.flagIdSuffix != "" {
 		if len(c.flagIdSuffix) != 10 {
 			c.UI.Error("Invalid ID suffix, must be exactly 10 characters")
@@ -702,9 +714,9 @@ func (c *Command) Run(args []string) int {
 	}
 
 	var opts []base.Option
-	if c.flagCreateLoopbackHostPlugin {
+	if c.flagCreateLoopbackPlugin {
 		c.DevLoopbackPluginId = "pl_1234567890"
-		c.EnabledPlugins = append(c.EnabledPlugins, base.EnabledPluginHostLoopback)
+		c.EnabledPlugins = append(c.EnabledPlugins, base.EnabledPluginLoopback)
 		c.Config.Controller.Scheduler.JobRunIntervalDuration = 100 * time.Millisecond
 	}
 	switch c.flagDatabaseUrl {
@@ -738,7 +750,7 @@ func (c *Command) Run(args []string) int {
 	}
 
 	{
-		c.EnabledPlugins = append(c.EnabledPlugins, base.EnabledPluginHostAws, base.EnabledPluginHostAzure)
+		c.EnabledPlugins = append(c.EnabledPlugins, base.EnabledPluginAws, base.EnabledPluginHostAzure)
 		conf := &controller.Config{
 			RawConfig: c.Config,
 			Server:    c.Server,
@@ -772,7 +784,7 @@ func (c *Command) Run(args []string) int {
 		}
 
 		var err error
-		c.worker, err = worker.New(conf)
+		c.worker, err = worker.New(c.Context, conf)
 		if err != nil {
 			c.UI.Error(fmt.Errorf("Error initializing worker: %w", err).Error())
 			return base.CommandCliError

internal/cmd/commands/server/server.go

@@ -470,8 +470,9 @@ func (c *Command) Run(args []string) int {
 		}
 	}
 
+	c.EnabledPlugins = append(c.EnabledPlugins, base.EnabledPluginAws)
 	if c.Config.Controller != nil {
-		c.EnabledPlugins = append(c.EnabledPlugins, base.EnabledPluginHostAws, base.EnabledPluginHostAzure)
+		c.EnabledPlugins = append(c.EnabledPlugins, base.EnabledPluginHostAzure)
 		if err := c.StartController(c.Context); err != nil {
 			c.UI.Error(err.Error())
 			return base.CommandCliError
@@ -624,7 +625,7 @@ func (c *Command) StartWorker() error {
 	}
 
 	var err error
-	c.worker, err = worker.New(conf)
+	c.worker, err = worker.New(c.Context, conf)
 	if err != nil {
 		return fmt.Errorf("Error initializing worker: %w", err)
 	}

internal/cmd/config/config.go

@@ -162,11 +162,11 @@ type Config struct {
 }
 
 type Controller struct {
-	Name              string     `hcl:"name"`
-	Description       string     `hcl:"description"`
-	Database          *Database  `hcl:"database"`
-	PublicClusterAddr string     `hcl:"public_cluster_addr"`
-	Scheduler         *Scheduler `hcl:"scheduler"`
+	Name              string    `hcl:"name"`
+	Description       string    `hcl:"description"`
+	Database          *Database `hcl:"database"`
+	PublicClusterAddr string    `hcl:"public_cluster_addr"`
+	Scheduler         Scheduler `hcl:"scheduler"`
 
 	// AuthTokenTimeToLive is the total valid lifetime of a token denoted by time.Duration
 	AuthTokenTimeToLive         any           `hcl:"auth_token_time_to_live"`
@@ -568,22 +568,20 @@ func Parse(d string) (*Config, error) {
 			result.Controller.GracefulShutdownWaitDuration = t
 		}
 
-		if result.Controller.Scheduler != nil {
-			if result.Controller.Scheduler.JobRunInterval != "" {
-				t, err := parseutil.ParseDurationSecond(result.Controller.Scheduler.JobRunInterval)
-				if err != nil {
-					return result, err
-				}
-				result.Controller.Scheduler.JobRunIntervalDuration = t
+		if result.Controller.Scheduler.JobRunInterval != "" {
+			t, err := parseutil.ParseDurationSecond(result.Controller.Scheduler.JobRunInterval)
+			if err != nil {
+				return result, err
 			}
+			result.Controller.Scheduler.JobRunIntervalDuration = t
+		}
 
-			if result.Controller.Scheduler.MonitorInterval != "" {
-				t, err := parseutil.ParseDurationSecond(result.Controller.Scheduler.MonitorInterval)
-				if err != nil {
-					return result, err
-				}
-				result.Controller.Scheduler.MonitorIntervalDuration = t
+		if result.Controller.Scheduler.MonitorInterval != "" {
+			t, err := parseutil.ParseDurationSecond(result.Controller.Scheduler.MonitorInterval)
+			if err != nil {
+				return result, err
 			}
+			result.Controller.Scheduler.MonitorIntervalDuration = t
 		}
 
 		workerStatusGracePeriod := result.Controller.WorkerStatusGracePeriod

internal/cmd/config/config_load_test.go

@@ -403,8 +403,11 @@ func TestLoad(t *testing.T) {
 						ConnMaxIdleTimeDuration:   nil,
 						SkipSharedLockAcquisition: false,
 					},
-					PublicClusterAddr:               "",
-					Scheduler:                       nil,
+					PublicClusterAddr: "",
+					Scheduler: config.Scheduler{
+						JobRunInterval:  nil,
+						MonitorInterval: nil,
+					},
 					AuthTokenTimeToLive:             nil,
 					AuthTokenTimeToLiveDuration:     0,
 					AuthTokenTimeToStale:            nil,
@@ -823,8 +826,11 @@ func TestLoad(t *testing.T) {
 						ConnMaxIdleTimeDuration:   nil,
 						SkipSharedLockAcquisition: false,
 					},
-					PublicClusterAddr:               "",
-					Scheduler:                       nil,
+					PublicClusterAddr: "",
+					Scheduler: config.Scheduler{
+						JobRunInterval:  nil,
+						MonitorInterval: nil,
+					},
 					AuthTokenTimeToLive:             nil,
 					AuthTokenTimeToLiveDuration:     0,
 					AuthTokenTimeToStale:            nil,
@@ -1238,8 +1244,11 @@ func TestLoad(t *testing.T) {
 						ConnMaxIdleTimeDuration:   nil,
 						SkipSharedLockAcquisition: false,
 					},
-					PublicClusterAddr:               "",
-					Scheduler:                       nil,
+					PublicClusterAddr: "",
+					Scheduler: config.Scheduler{
+						JobRunInterval:  nil,
+						MonitorInterval: nil,
+					},
 					AuthTokenTimeToLive:             nil,
 					AuthTokenTimeToLiveDuration:     0,
 					AuthTokenTimeToStale:            nil,
@@ -1653,8 +1662,11 @@ func TestLoad(t *testing.T) {
 						ConnMaxIdleTimeDuration:   nil,
 						SkipSharedLockAcquisition: false,
 					},
-					PublicClusterAddr:               "",
-					Scheduler:                       nil,
+					PublicClusterAddr: "",
+					Scheduler: config.Scheduler{
+						JobRunInterval:  nil,
+						MonitorInterval: nil,
+					},
 					AuthTokenTimeToLive:             nil,
 					AuthTokenTimeToLiveDuration:     0,
 					AuthTokenTimeToStale:            nil,
@@ -1739,8 +1751,11 @@ func TestLoad(t *testing.T) {
 						ConnMaxIdleTimeDuration:   nil,
 						SkipSharedLockAcquisition: false,
 					},
-					PublicClusterAddr:               "",
-					Scheduler:                       nil,
+					PublicClusterAddr: "",
+					Scheduler: config.Scheduler{
+						JobRunInterval:  nil,
+						MonitorInterval: nil,
+					},
 					AuthTokenTimeToLive:             nil,
 					AuthTokenTimeToLiveDuration:     0,
 					AuthTokenTimeToStale:            nil,

internal/daemon/controller/controller.go

@@ -45,7 +45,7 @@ import (
 	"github.com/hashicorp/boundary/internal/types/scope"
 	host_plugin_assets "github.com/hashicorp/boundary/plugins/host"
 	"github.com/hashicorp/boundary/sdk/pbs/plugin"
-	external_host_plugins "github.com/hashicorp/boundary/sdk/plugins/host"
+	external_plugins "github.com/hashicorp/boundary/sdk/plugins"
 	"github.com/hashicorp/boundary/version"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-secure-stdlib/mlock"
@@ -251,29 +251,30 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
 			}
 		}
 		switch enabledPlugin {
-		case base.EnabledPluginHostLoopback:
+		case base.EnabledPluginLoopback:
 			lp, err := loopback.NewLoopbackPlugin()
 			if err != nil {
 				return nil, fmt.Errorf("error creating loopback plugin: %w", err)
 			}
 			plg := loopback.NewWrappingPluginHostClient(lp)
 			opts := []host.Option{
-				host.WithDescription("Provides an initial loopback host plugin in Boundary"),
+				host.WithDescription("Provides an initial loopback storage and host plugin in Boundary"),
 				host.WithPublicId(conf.DevLoopbackPluginId),
 			}
+			// TODO(storage): Register loopback as storage supported
 			if _, err = conf.RegisterHostPlugin(ctx, "loopback", plg, opts...); err != nil {
 				return nil, err
 			}
-		case base.EnabledPluginHostAzure, base.EnabledPluginHostAws:
+		case base.EnabledPluginHostAzure, base.EnabledPluginAws:
 			pluginType := strings.ToLower(enabledPlugin.String())
-			client, cleanup, err := external_host_plugins.CreateHostPlugin(
+			client, cleanup, err := external_plugins.CreateHostPlugin(
 				ctx,
 				pluginType,
-				external_host_plugins.WithPluginOptions(
+				external_plugins.WithPluginOptions(
 					pluginutil.WithPluginExecutionDirectory(conf.RawConfig.Plugins.ExecutionDir),
 					pluginutil.WithPluginsFilesystem(host_plugin_assets.HostPluginPrefix, host_plugin_assets.FileSystem()),
 				),
-				external_host_plugins.WithLogger(pluginLogger.Named(pluginType)),
+				external_plugins.WithLogger(pluginLogger.Named(pluginType)),
 			)
 			if err != nil {
 				return nil, fmt.Errorf("error creating %s host plugin: %w", pluginType, err)
@@ -337,14 +338,13 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
 	}
 	// TODO: Allow setting run jobs limit from config
 	schedulerOpts := []scheduler.Option{scheduler.WithRunJobsLimit(-1)}
-	if sche := c.conf.RawConfig.Controller.Scheduler; sche != nil {
-		if sche.JobRunIntervalDuration > 0 {
-			schedulerOpts = append(schedulerOpts, scheduler.WithRunJobsInterval(sche.JobRunIntervalDuration))
-		}
-		if sche.MonitorIntervalDuration > 0 {
-			schedulerOpts = append(schedulerOpts, scheduler.WithMonitorInterval(sche.MonitorIntervalDuration))
-		}
+	if c.conf.RawConfig.Controller.Scheduler.JobRunIntervalDuration > 0 {
+		schedulerOpts = append(schedulerOpts, scheduler.WithRunJobsInterval(c.conf.RawConfig.Controller.Scheduler.JobRunIntervalDuration))
 	}
+	if c.conf.RawConfig.Controller.Scheduler.MonitorIntervalDuration > 0 {
+		schedulerOpts = append(schedulerOpts, scheduler.WithMonitorInterval(c.conf.RawConfig.Controller.Scheduler.MonitorIntervalDuration))
+	}
+
 	c.scheduler, err = scheduler.New(c.conf.RawConfig.Controller.Name, jobRepoFn, schedulerOpts...)
 	if err != nil {
 		return nil, fmt.Errorf("error creating new scheduler: %w", err)

internal/daemon/controller/controller_test.go

@@ -255,7 +255,7 @@ func TestController_NewPluginsConfig(t *testing.T) {
 	initialConfig.Plugins.ExecutionDir = tmpDir
 	conf := TestControllerConfig(t, ctx, tc, &TestControllerOpts{Config: initialConfig})
 	conf.EnabledPlugins = []base.EnabledPlugin{
-		base.EnabledPluginHostAws,
+		base.EnabledPluginAws,
 		base.EnabledPluginHostAzure,
 	}
 

internal/daemon/controller/testing.go

@@ -614,7 +614,7 @@ func TestControllerConfig(t testing.TB, ctx context.Context, tc *TestController,
 	tc.b.DevUnprivilegedOidcAccountId = DefaultTestUnprivilegedOidcAccountId
 	tc.b.DevLoopbackPluginId = DefaultTestPluginId
 
-	tc.b.EnabledPlugins = append(tc.b.EnabledPlugins, base.EnabledPluginHostLoopback)
+	tc.b.EnabledPlugins = append(tc.b.EnabledPlugins, base.EnabledPluginLoopback)
 
 	// Start a logger
 	tc.b.Logger = opts.Logger
@@ -632,10 +632,6 @@ func TestControllerConfig(t testing.TB, ctx context.Context, tc *TestController,
 	if opts.Config.Controller.Name == "" {
 		require.NoError(t, opts.Config.Controller.InitNameIfEmpty())
 	}
-
-	if opts.Config.Controller.Scheduler == nil {
-		opts.Config.Controller.Scheduler = new(config.Scheduler)
-	}
 	opts.Config.Controller.Scheduler.JobRunIntervalDuration = opts.SchedulerRunJobInterval
 
 	if opts.EnableEventing {

internal/daemon/worker/listeners_test.go

@@ -78,9 +78,10 @@ func TestStartListeners(t *testing.T) {
 			err := conf.SetupListeners(nil, conf.RawConfig.SharedConfig, []string{"proxy"})
 			require.NoError(t, err)
 
-			w, err := New(conf)
+			ctx := context.Background()
+			w, err := New(ctx, conf)
 			require.NoError(t, err)
-			w.baseContext = context.Background()
+			w.baseContext = ctx
 
 			var dummyClientProducer handlers.UpstreamMessageServiceClientProducer
 			dummyClientProducer = func(ctx context.Context) (pbs.UpstreamMessageServiceClient, error) {

internal/daemon/worker/testing.go

@@ -311,6 +311,8 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker {
 	if opts.WorkerRecordingStoragePath != "" {
 		opts.Config.Worker.RecordingStoragePath = opts.WorkerRecordingStoragePath
 	}
+
+	tw.b.EnabledPlugins = append(tw.b.EnabledPlugins, base.EnabledPluginLoopback)
 	tw.name = opts.Config.Worker.Name
 
 	if opts.SuccessfulStatusGracePeriodDuration != 0 {
@@ -356,7 +358,7 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker {
 		Server:    tw.b,
 	}
 
-	tw.w, err = New(conf)
+	tw.w, err = New(ctx, conf)
 	if err != nil {
 		tw.Shutdown()
 		t.Fatal(err)

internal/daemon/worker/worker.go

@@ -31,10 +31,13 @@ import (
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/server"
 	"github.com/hashicorp/boundary/internal/storage"
+	host_plugin_assets "github.com/hashicorp/boundary/plugins/host"
 	plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	external_plugins "github.com/hashicorp/boundary/sdk/plugins"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-secure-stdlib/base62"
 	"github.com/hashicorp/go-secure-stdlib/mlock"
+	"github.com/hashicorp/go-secure-stdlib/pluginutil/v2"
 	"github.com/hashicorp/go-secure-stdlib/strutil"
 	"github.com/hashicorp/nodeenrollment"
 	nodeenet "github.com/hashicorp/nodeenrollment/net"
@@ -80,7 +83,7 @@ type recorderCache any
 // create its reverseConnReceiver
 var reverseConnReceiverFactory func() reverseConnReceiver
 
-var recordingStorageFactory func(ctx context.Context, path string, plgClients map[string]plgpb.StoragePluginServiceClient) (storage.RecordingStorage, error)
+var recordingStorageFactory func(ctx context.Context, path string, plgClients map[string]plgpb.StoragePluginServiceClient, enableLoopback bool) (storage.RecordingStorage, error)
 
 var recorderCacheFactory func() recorderCache
 
@@ -148,8 +151,6 @@ type Worker struct {
 	// The storage for session recording
 	RecordingStorage storage.RecordingStorage
 
-	enabledPlugins []base.EnabledPlugin
-
 	// downstream workers and routes to those workers
 	downstreamWorkers  downstreamers
 	downstreamReceiver reverseConnReceiver
@@ -173,7 +174,7 @@ type Worker struct {
 	pkiConnManager *cluster.DownstreamManager
 }
 
-func New(conf *Config) (*Worker, error) {
+func New(ctx context.Context, conf *Config) (*Worker, error) {
 	const op = "worker.New"
 	metric.InitializeHttpCollectors(conf.PrometheusRegisterer)
 	metric.InitializeWebsocketCollectors(conf.PrometheusRegisterer)
@@ -217,15 +218,37 @@ func New(conf *Config) (*Worker, error) {
 	}
 
 	if w.conf.RawConfig.Worker.RecordingStoragePath != "" && recordingStorageFactory != nil {
+		pluginLogger, err := event.NewHclogLogger(ctx, w.conf.Server.Eventer)
+		if err != nil {
+			return nil, fmt.Errorf("error creating storage catalog plugin logger: %w", err)
+		}
 		plgClients := make(map[string]plgpb.StoragePluginServiceClient)
-		for _, enabledPlugin := range w.enabledPlugins {
+		var enableStorageLoopback bool
+
+		for _, enabledPlugin := range w.conf.Server.EnabledPlugins {
 			switch enabledPlugin {
-			// TODO(storage): Add mock plugin when enabled, also make sure to enable in dev mode
-			// TODO(storage): Add AWS storage plugin
+			case base.EnabledPluginAws:
+				pluginType := strings.ToLower(enabledPlugin.String())
+				client, cleanup, err := external_plugins.CreateStoragePlugin(
+					ctx,
+					pluginType,
+					external_plugins.WithPluginOptions(
+						pluginutil.WithPluginExecutionDirectory(conf.RawConfig.Plugins.ExecutionDir),
+						pluginutil.WithPluginsFilesystem(host_plugin_assets.HostPluginPrefix, host_plugin_assets.FileSystem()),
+					),
+					external_plugins.WithLogger(pluginLogger.Named(pluginType)),
+				)
+				if err != nil {
+					return nil, fmt.Errorf("error creating %s storage plugin: %w", pluginType, err)
+				}
+				conf.ShutdownFuncs = append(conf.ShutdownFuncs, cleanup)
+				plgClients[pluginType] = client
+			case base.EnabledPluginLoopback:
+				enableStorageLoopback = true
 			}
 		}
 
-		s, err := recordingStorageFactory(w.baseContext, w.conf.RawConfig.Worker.RecordingStoragePath, plgClients)
+		s, err := recordingStorageFactory(ctx, w.conf.RawConfig.Worker.RecordingStoragePath, plgClients, enableStorageLoopback)
 		if err != nil {
 			return nil, fmt.Errorf("error create recording storage: %w", err)
 		}

internal/daemon/worker/worker_test.go

@@ -129,7 +129,7 @@ func TestWorkerNew(t *testing.T) {
 			tt.in.Logger = hclog.Default()
 			tt.in.RawConfig = &config.Config{SharedConfig: &configutil.SharedConfig{DisableMlock: true}}
 
-			w, err := New(tt.in)
+			w, err := New(context.Background(), tt.in)
 			if tt.expErr {
 				require.EqualError(t, err, tt.expErrMsg)
 				require.Nil(t, w)
@@ -286,7 +286,7 @@ func Test_Worker_getSessionTls(t *testing.T) {
 		},
 	}
 	conf.RawConfig = &config.Config{SharedConfig: &configutil.SharedConfig{DisableMlock: true}}
-	w, err := New(conf)
+	w, err := New(context.Background(), conf)
 	require.NoError(t, err)
 	w.lastStatusSuccess.Store(&LastStatusInformation{StatusResponse: &services.StatusResponse{}, StatusTime: time.Now(), LastCalculatedUpstreams: nil})
 	w.baseContext = context.Background()

internal/plugin/loopback/loopback.go

@@ -36,7 +36,10 @@ type LoopbackPlugin struct {
 	*LoopbackStorage
 }
 
-// NewLoopbackPlugin returns a new loopback plugin
+// NewLoopbackPlugin returns a new loopback pluginre
+// For storage service testings NewLoopbackPlugin Supports WithMockErrors
+// and WithMockBuckets as options. If no mock buckets are provided,
+// a bucket named `default` will be created.
 func NewLoopbackPlugin(opt ...TestOption) (*LoopbackPlugin, error) {
 	opts, err := getTestOpts(opt...)
 	if err != nil {
@@ -49,7 +52,7 @@ func NewLoopbackPlugin(opt ...TestOption) (*LoopbackPlugin, error) {
 			hostMap: make(map[string][]*loopbackPluginHostInfo),
 		},
 		LoopbackStorage: &LoopbackStorage{
-			buckets: make(map[BucketName]Bucket),
+			buckets: map[BucketName]Bucket{"default": {}},
 			errs:    make([]PluginMockError, 0),
 		},
 	}

sdk/plugins/load.go

@@ -1,12 +1,11 @@
 // Copyright (c) HashiCorp, Inc.
 // SPDX-License-Identifier: MPL-2.0
 
-package external_host_plugins
+package external_plugins
 
 import (
 	"context"
 	"fmt"
-	"io/fs"
 	"strings"
 
 	pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
@@ -14,25 +13,49 @@ import (
 	"github.com/hashicorp/go-secure-stdlib/pluginutil/v2"
 )
 
-// NOTE: This package could probably use some reflect based bits to allow
-// loading other types of plugins later. That is an exercise left for future
-// refactoring, but worth calling out now.
-
-type pluginInfo struct {
-	containerFs  fs.FS
-	filename     string
-	creationFunc func() (pb.HostPluginServiceClient, error)
-}
-
 // CreateHostPlugin takes in a type, parses the various options to look for a
 // plugin matching that name, and returns a host plugin client, a cleanup
 // function to execute on shutdown of the enclosing program, and an error.
-func CreateHostPlugin(
+func CreateHostPlugin(ctx context.Context, pluginType string, opt ...Option) (pb.HostPluginServiceClient, func() error, error) {
+	raw, cleanup, err := createPlugin(ctx, pluginType, hostServicePluginSetName, opt...)
+	if err != nil {
+		return nil, cleanup, err
+	}
+
+	var ok bool
+	hp, ok := raw.(pb.HostPluginServiceClient)
+	if !ok {
+		return nil, cleanup, fmt.Errorf("error converting rpc storage plugin of type %T to normal wrapper", raw)
+	}
+
+	return hp, cleanup, nil
+}
+
+// CreateStoragePlugin takes in a type, parses the various options to look for a
+// plugin matching that name, and returns a storage plugin client, a cleanup
+// function to execute on shutdown of the enclosing program, and an error.
+func CreateStoragePlugin(ctx context.Context, pluginType string, opt ...Option) (pb.StoragePluginServiceClient, func() error, error) {
+	raw, cleanup, err := createPlugin(ctx, pluginType, storageServicePluginSetName, opt...)
+	if err != nil {
+		return nil, cleanup, err
+	}
+
+	var ok bool
+	sp, ok := raw.(pb.StoragePluginServiceClient)
+	if !ok {
+		return nil, cleanup, fmt.Errorf("error converting rpc storage plugin of type %T to normal wrapper", raw)
+	}
+
+	return sp, cleanup, nil
+}
+
+func createPlugin(
 	ctx context.Context,
 	pluginType string,
+	pluginSetName string,
 	opt ...Option,
 ) (
-	hp pb.HostPluginServiceClient,
+	raw any,
 	cleanup func() error,
 	retErr error,
 ) {
@@ -46,7 +69,7 @@ func CreateHostPlugin(
 
 	opts, err := getOpts(opt...)
 	if err != nil {
-		return nil, nil, fmt.Errorf("error parsing host plugin options: %w", err)
+		return nil, nil, fmt.Errorf("error parsing plugin options: %w", err)
 	}
 
 	// First, scan available plugins, then find the right one to use
@@ -55,7 +78,7 @@ func CreateHostPlugin(
 			opts.withPluginOptions,
 			pluginutil.WithPluginClientCreationFunc(
 				func(pluginPath string, _ ...pluginutil.Option) (*plugin.Client, error) {
-					return NewHostPluginClient(pluginPath, WithLogger(opts.withLogger))
+					return NewPluginClient(pluginPath, pluginSetName, WithLogger(opts.withLogger))
 				}),
 		)...)
 	if err != nil {
@@ -68,22 +91,15 @@ func CreateHostPlugin(
 		return nil, cleanup, err
 	}
 
-	var raw any
 	switch client := plugClient.(type) {
 	case plugin.ClientProtocol:
-		raw, err = client.Dispense(hostServicePluginSetName)
+		raw, err = client.Dispense(pluginSetName)
 		if err != nil {
-			return nil, cleanup, fmt.Errorf("error dispensing host plugin: %w", err)
+			return nil, cleanup, fmt.Errorf("error dispensing %q plugin: %w", pluginSetName, err)
 		}
 	default:
 		return nil, cleanup, fmt.Errorf("unable to understand type %T of raw plugin", raw)
 	}
 
-	var ok bool
-	hp, ok = raw.(pb.HostPluginServiceClient)
-	if !ok {
-		return nil, cleanup, fmt.Errorf("error converting rpc host plugin of type %T to normal wrapper", raw)
-	}
-
-	return hp, cleanup, nil
+	return raw, cleanup, nil
 }

sdk/plugins/options.go

@@ -1,7 +1,7 @@
 // Copyright (c) HashiCorp, Inc.
 // SPDX-License-Identifier: MPL-2.0
 
-package external_host_plugins
+package external_plugins
 
 import (
 	"fmt"

sdk/plugins/plugin.go

@@ -1,10 +1,11 @@
 // Copyright (c) HashiCorp, Inc.
 // SPDX-License-Identifier: MPL-2.0
 
-package external_host_plugins
+package external_plugins
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"os/exec"
@@ -17,19 +18,20 @@ import (
 )
 
 const (
-	hostServicePluginSetName = "host-plugin"
+	hostServicePluginSetName    = "host-plugin"
+	storageServicePluginSetName = "storage-plugin"
 )
 
 // HandshakeConfig is a shared config that can be used regardless of plugin, to
 // avoid having to know type-specific things about each plugin
 var HandshakeConfig = plugin.HandshakeConfig{
-	MagicCookieKey:   "HASHICORP_BOUNDARY_HOST_PLUGIN",
-	MagicCookieValue: hostServicePluginSetName,
+	MagicCookieKey:   "HASHICORP_BOUNDARY_PLUGIN",
+	MagicCookieValue: "boundary-plugin",
 }
 
 // ServeHostPlugin is a generic function to start serving a host plugin service as a
 // plugin
-func ServeHostPlugin(svc pb.HostPluginServiceServer, opt ...Option) error {
+func ServePlugin(svc any, opt ...Option) error {
 	opts, err := getOpts(opt...)
 	if err != nil {
 		return err
@@ -43,14 +45,29 @@ func ServeHostPlugin(svc pb.HostPluginServiceServer, opt ...Option) error {
 		}
 	}()
 
-	hostServiceServer, err := NewHostPluginServiceServer(svc)
-	if err != nil {
-		return err
+	plugins := make(map[string]plugin.Plugin)
+	if hostSvc, ok := svc.(pb.HostPluginServiceServer); ok {
+		hostServiceServer, err := NewHostPluginServiceServer(hostSvc)
+		if err != nil {
+			return err
+		}
+		plugins[hostServicePluginSetName] = hostServiceServer
 	}
+	if storageSvc, ok := svc.(pb.StoragePluginServiceServer); ok {
+		storageServiceServer, err := NewStoragePluginServiceServer(storageSvc)
+		if err != nil {
+			return err
+		}
+		plugins[storageServicePluginSetName] = storageServiceServer
+	}
+	if len(plugins) == 0 {
+		return errors.New("no valid plugin server provided")
+	}
+
 	plugin.Serve(&plugin.ServeConfig{
 		HandshakeConfig: HandshakeConfig,
 		VersionedPlugins: map[int]plugin.PluginSet{
-			1: {hostServicePluginSetName: hostServiceServer},
+			1: plugins,
 		},
 		Logger:     opts.withLogger,
 		GRPCServer: plugin.DefaultGRPCServer,
@@ -58,6 +75,34 @@ func ServeHostPlugin(svc pb.HostPluginServiceServer, opt ...Option) error {
 	return nil
 }
 
+func NewPluginClient(pluginPath, setName string, opt ...Option) (*plugin.Client, error) {
+	opts, err := getOpts(opt...)
+	if err != nil {
+		return nil, err
+	}
+
+	var set plugin.PluginSet
+	switch setName {
+	case hostServicePluginSetName:
+		set = plugin.PluginSet{hostServicePluginSetName: &hostPlugin{}}
+	case storageServicePluginSetName:
+		set = plugin.PluginSet{storageServicePluginSetName: &storagePlugin{}}
+	}
+
+	return plugin.NewClient(&plugin.ClientConfig{
+		HandshakeConfig: HandshakeConfig,
+		VersionedPlugins: map[int]plugin.PluginSet{
+			1: set,
+		},
+		Cmd: exec.Command(pluginPath),
+		AllowedProtocols: []plugin.Protocol{
+			plugin.ProtocolGRPC,
+		},
+		Logger:   opts.withLogger,
+		AutoMTLS: true,
+	}), nil
+}
+
 type hostPlugin struct {
 	plugin.Plugin
 
@@ -73,27 +118,6 @@ func NewHostPluginServiceServer(impl pb.HostPluginServiceServer) (*hostPlugin, e
 	}, nil
 }
 
-func NewHostPluginClient(pluginPath string, opt ...Option) (*plugin.Client, error) {
-	opts, err := getOpts(opt...)
-	if err != nil {
-		return nil, err
-	}
-	hostServiceClient := &hostPlugin{}
-
-	return plugin.NewClient(&plugin.ClientConfig{
-		HandshakeConfig: HandshakeConfig,
-		VersionedPlugins: map[int]plugin.PluginSet{
-			1: {hostServicePluginSetName: hostServiceClient},
-		},
-		Cmd: exec.Command(pluginPath),
-		AllowedProtocols: []plugin.Protocol{
-			plugin.ProtocolGRPC,
-		},
-		Logger:   opts.withLogger,
-		AutoMTLS: true,
-	}), nil
-}
-
 func (h *hostPlugin) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) error {
 	pb.RegisterHostPluginServiceServer(s, h.impl)
 	return nil
@@ -102,3 +126,27 @@ func (h *hostPlugin) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) error
 func (h *hostPlugin) GRPCClient(ctx context.Context, broker *plugin.GRPCBroker, c *grpc.ClientConn) (any, error) {
 	return pb.NewHostPluginServiceClient(c), nil
 }
+
+type storagePlugin struct {
+	plugin.Plugin
+
+	impl pb.StoragePluginServiceServer
+}
+
+func NewStoragePluginServiceServer(impl pb.StoragePluginServiceServer) (*storagePlugin, error) {
+	if impl == nil {
+		return nil, fmt.Errorf("empty underlying storage plugin passed in")
+	}
+	return &storagePlugin{
+		impl: impl,
+	}, nil
+}
+
+func (p *storagePlugin) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) error {
+	pb.RegisterStoragePluginServiceServer(s, p.impl)
+	return nil
+}
+
+func (p *storagePlugin) GRPCClient(ctx context.Context, broker *plugin.GRPCBroker, c *grpc.ClientConn) (any, error) {
+	return pb.NewStoragePluginServiceClient(c), nil
+}