diff --git a/internal/cmd/base/base.go b/internal/cmd/base/base.go index b96c4982e..6cf3a91a1 100644 --- a/internal/cmd/base/base.go +++ b/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" diff --git a/internal/cmd/commands/database/migrate.go b/internal/cmd/commands/database/migrate.go index c1eda3990..9088f8f7a 100644 --- a/internal/cmd/commands/database/migrate.go +++ b/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()) diff --git a/internal/cmd/commands/dev/dev.go b/internal/cmd/commands/dev/dev.go index 596ddfd3e..c0da07c73 100644 --- a/internal/cmd/commands/dev/dev.go +++ b/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 diff --git a/internal/cmd/commands/server/server.go b/internal/cmd/commands/server/server.go index 4c72cf202..0774b53ea 100644 --- a/internal/cmd/commands/server/server.go +++ b/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) } diff --git a/internal/cmd/config/config.go b/internal/cmd/config/config.go index 9e1d3e72d..b17cbeb8b 100644 --- a/internal/cmd/config/config.go +++ b/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 diff --git a/internal/cmd/config/config_load_test.go b/internal/cmd/config/config_load_test.go index d4ac9928a..d0c24a769 100644 --- a/internal/cmd/config/config_load_test.go +++ b/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, diff --git a/internal/daemon/controller/controller.go b/internal/daemon/controller/controller.go index 3ca9eebdc..27d562a9c 100644 --- a/internal/daemon/controller/controller.go +++ b/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) diff --git a/internal/daemon/controller/controller_test.go b/internal/daemon/controller/controller_test.go index 2f85393e2..a495ff3de 100644 --- a/internal/daemon/controller/controller_test.go +++ b/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, } diff --git a/internal/daemon/controller/testing.go b/internal/daemon/controller/testing.go index 4a5383240..0d7420839 100644 --- a/internal/daemon/controller/testing.go +++ b/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 { diff --git a/internal/daemon/worker/listeners_test.go b/internal/daemon/worker/listeners_test.go index 36d874df1..ce889745d 100644 --- a/internal/daemon/worker/listeners_test.go +++ b/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) { diff --git a/internal/daemon/worker/testing.go b/internal/daemon/worker/testing.go index 32d5688cd..f3dd5c0c5 100644 --- a/internal/daemon/worker/testing.go +++ b/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) diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index 8c495258c..3a50f167d 100644 --- a/internal/daemon/worker/worker.go +++ b/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) } diff --git a/internal/daemon/worker/worker_test.go b/internal/daemon/worker/worker_test.go index ae10b14b2..ab92a1502 100644 --- a/internal/daemon/worker/worker_test.go +++ b/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() diff --git a/internal/plugin/loopback/loopback.go b/internal/plugin/loopback/loopback.go index 3db22886e..c45821f99 100644 --- a/internal/plugin/loopback/loopback.go +++ b/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), }, } diff --git a/sdk/plugins/host/load.go b/sdk/plugins/load.go similarity index 50% rename from sdk/plugins/host/load.go rename to sdk/plugins/load.go index 5f491b102..b22bb93f1 100644 --- a/sdk/plugins/host/load.go +++ b/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 } diff --git a/sdk/plugins/host/options.go b/sdk/plugins/options.go similarity index 97% rename from sdk/plugins/host/options.go rename to sdk/plugins/options.go index 2e8ca359c..a7f96cffa 100644 --- a/sdk/plugins/host/options.go +++ b/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" diff --git a/sdk/plugins/host/plugin.go b/sdk/plugins/plugin.go similarity index 50% rename from sdk/plugins/host/plugin.go rename to sdk/plugins/plugin.go index 722d71573..d5526e07e 100644 --- a/sdk/plugins/host/plugin.go +++ b/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 +}