From 11124794bf930a4826b30b1364d9fdfd17767094 Mon Sep 17 00:00:00 2001 From: Jim Date: Sun, 25 Jul 2021 17:40:54 -0400 Subject: [PATCH] Events: add node for formatting/filtering cloudevents (#1390) --- internal/cmd/base/event_flags.go | 79 +++- internal/cmd/base/event_flags_test.go | 154 ++++++++ internal/cmd/base/option.go | 3 +- internal/cmd/base/option_test.go | 1 + internal/cmd/base/server_test.go | 6 +- internal/cmd/base/servers.go | 27 +- internal/cmd/commands/database/init.go | 13 +- internal/cmd/commands/database/migrate.go | 14 +- internal/cmd/commands/dev/dev.go | 66 +++- internal/cmd/commands/server/server.go | 18 +- internal/cmd/commands/server/server_test.go | 2 +- internal/cmd/config/config.go | 31 ++ internal/filter/filter.go | 43 +++ internal/filter/filter_test.go | 53 +++ internal/observability/event/context.go | 24 +- internal/observability/event/context_test.go | 222 ++++++++--- internal/observability/event/event.go | 5 +- internal/observability/event/event_audit.go | 2 +- internal/observability/event/event_error.go | 2 +- .../observability/event/event_observation.go | 72 +++- .../event/event_observation_test.go | 16 +- internal/observability/event/event_sys.go | 2 +- .../observability/event/event_sys_test.go | 14 + internal/observability/event/eventer.go | 109 ++++-- .../observability/event/eventer_retry_test.go | 2 +- internal/observability/event/eventer_test.go | 186 +++++++--- internal/observability/event/id.go | 6 +- internal/observability/event/id_test.go | 2 +- internal/observability/event/node.go | 126 +++++++ internal/observability/event/node_test.go | 346 ++++++++++++++++++ internal/observability/event/options.go | 25 ++ internal/observability/event/options_test.go | 27 ++ internal/observability/event/sink_config.go | 34 +- internal/observability/event/sink_format.go | 5 +- internal/observability/event/testing.go | 1 + internal/servers/common/handler.go | 7 + internal/servers/common/handler_test.go | 62 ++-- internal/servers/controller/controller.go | 8 +- .../servers/controller/handlers/filtering.go | 42 +-- .../controller/handlers/filtering_test.go | 18 - internal/servers/controller/testing.go | 23 +- internal/servers/worker/testing.go | 6 +- internal/servers/worker/worker.go | 7 +- 43 files changed, 1591 insertions(+), 320 deletions(-) create mode 100644 internal/filter/filter.go create mode 100644 internal/filter/filter_test.go create mode 100644 internal/observability/event/event_sys_test.go create mode 100644 internal/observability/event/node.go create mode 100644 internal/observability/event/node_test.go diff --git a/internal/cmd/base/event_flags.go b/internal/cmd/base/event_flags.go index 5eb690758..ba9d5b185 100644 --- a/internal/cmd/base/event_flags.go +++ b/internal/cmd/base/event_flags.go @@ -1,6 +1,12 @@ package base -import "github.com/hashicorp/boundary/internal/observability/event" +import ( + "fmt" + "strings" + + "github.com/hashicorp/boundary/internal/observability/event" + "github.com/hashicorp/go-bexpr" +) // EventFlags represent the cmd flags supported overriding the configured or // default event configuration @@ -8,6 +14,9 @@ type EventFlags struct { Format event.SinkFormat AuditEnabled *bool ObservationsEnabled *bool + SysEventsEnabled *bool + AllowFilters []string + DenyFilters []string } // Validate simply validates the flags @@ -16,6 +25,74 @@ func (ef *EventFlags) Validate() error { if err := ef.Format.Validate(); err != nil { return err } + for i, f := range ef.AllowFilters { + _, err := bexpr.CreateEvaluator(f) + if err != nil { + return fmt.Errorf("invalid allow filter '%s': %w", ef.AllowFilters[i], err) + } + } + for i, f := range ef.DenyFilters { + _, err := bexpr.CreateEvaluator(f) + if err != nil { + return fmt.Errorf("invalid deny filter '%s': %w", ef.DenyFilters[i], err) + } + } } + return nil } + +type ComposedOfEventArgs struct { + Format string + Observations string + Audit string + SysEvents string + Allow []string + Deny []string +} + +// NewEventFlags will create a new EventFlags based on the ComposedOfEventArgs +// which should be populated with command flags which have already been "parsed" +func NewEventFlags(defaultFormat event.SinkFormat, c ComposedOfEventArgs) (*EventFlags, error) { + const op = "base.NewEventFlags" + if defaultFormat == "" { + return nil, fmt.Errorf("%s: missing default sink format", op) + } + setTrue := true + setFalse := false + f := &EventFlags{ + Format: defaultFormat, + } + if c.Format != "" { + f.Format = event.SinkFormat(c.Format) + } + switch strings.ToLower(c.Observations) { + case "true": + f.ObservationsEnabled = &setTrue + case "false": + f.ObservationsEnabled = &setFalse + } + switch strings.ToLower(c.Audit) { + case "true": + f.AuditEnabled = &setTrue + case "false": + f.AuditEnabled = &setFalse + } + switch strings.ToLower(c.SysEvents) { + case "true": + f.SysEventsEnabled = &setTrue + case "false": + f.SysEventsEnabled = &setFalse + } + + if len(c.Allow) > 0 { + f.AllowFilters = c.Allow + } + if len(c.Deny) > 0 { + f.DenyFilters = c.Deny + } + if err := f.Validate(); err != nil { + return nil, fmt.Errorf("%s: %w", op, err) + } + return f, nil +} diff --git a/internal/cmd/base/event_flags_test.go b/internal/cmd/base/event_flags_test.go index 45e4f9d88..e207e2f27 100644 --- a/internal/cmd/base/event_flags_test.go +++ b/internal/cmd/base/event_flags_test.go @@ -22,6 +22,12 @@ func TestEventFlags_Validate(t *testing.T) { Format: event.JSONSinkFormat, }, }, + { + name: "valid-TextFormat", + flags: EventFlags{ + Format: event.TextSinkFormat, + }, + }, { name: "empty", flags: EventFlags{}, @@ -57,3 +63,151 @@ func TestEventFlags_Validate(t *testing.T) { }) } } + +func Test_NewEventFlags(t *testing.T) { + setTrue := true + setFalse := false + tests := []struct { + name string + defaultFormat event.SinkFormat + composedOf ComposedOfEventArgs + wantFlags *EventFlags + wantErr bool + wantErrIs error + wantErrContains string + }{ + { + name: "missing-default-format", + wantErr: true, + wantErrContains: "missing default sink format", + }, + { + name: "invalid-default-format", + defaultFormat: "invalid-format", + wantErr: true, + wantErrIs: event.ErrInvalidParameter, + wantErrContains: "'invalid-format' is not a valid sink format", + }, + { + name: "defaults", + defaultFormat: "cloudevents-json", + wantFlags: &EventFlags{ + Format: "cloudevents-json", + }, + }, + { + name: "override-format", + defaultFormat: "cloudevents-json", + composedOf: ComposedOfEventArgs{Format: "cloudevents-text"}, + wantFlags: &EventFlags{ + Format: "cloudevents-text", + }, + }, + { + name: "observations-true", + defaultFormat: "cloudevents-json", + composedOf: ComposedOfEventArgs{Format: "cloudevents-json", Observations: "true"}, + wantFlags: &EventFlags{ + Format: "cloudevents-json", + ObservationsEnabled: &setTrue, + }, + }, + { + name: "observations-false", + defaultFormat: "cloudevents-json", + composedOf: ComposedOfEventArgs{Format: "cloudevents-json", Observations: "false"}, + wantFlags: &EventFlags{ + Format: "cloudevents-json", + ObservationsEnabled: &setFalse, + }, + }, + { + name: "audit-true", + defaultFormat: "cloudevents-json", + composedOf: ComposedOfEventArgs{Format: "cloudevents-json", Audit: "true"}, + wantFlags: &EventFlags{ + Format: "cloudevents-json", + AuditEnabled: &setTrue, + }, + }, + { + name: "audit-false", + defaultFormat: "cloudevents-json", + composedOf: ComposedOfEventArgs{Format: "cloudevents-json", Audit: "false"}, + wantFlags: &EventFlags{ + Format: "cloudevents-json", + AuditEnabled: &setFalse, + }, + }, + { + name: "sysevents-true", + defaultFormat: "cloudevents-json", + composedOf: ComposedOfEventArgs{Format: "cloudevents-json", SysEvents: "true"}, + wantFlags: &EventFlags{ + Format: "cloudevents-json", + SysEventsEnabled: &setTrue, + }, + }, + { + name: "sysevents-false", + defaultFormat: "cloudevents-json", + composedOf: ComposedOfEventArgs{Format: "cloudevents-json", SysEvents: "false"}, + wantFlags: &EventFlags{ + Format: "cloudevents-json", + SysEventsEnabled: &setFalse, + }, + }, + { + name: "valid-allow", + defaultFormat: "cloudevents-json", + composedOf: ComposedOfEventArgs{Format: "cloudevents-json", Allow: []string{`"/Data/Header/status" == 401`}}, + wantFlags: &EventFlags{ + Format: "cloudevents-json", + AllowFilters: []string{`"/Data/Header/status" == 401`}, + }, + }, + { + name: "invalid-allow", + defaultFormat: "cloudevents-json", + composedOf: ComposedOfEventArgs{Format: "cloudevents-json", Allow: []string{`"/Data/Header/status" $$== 401`}}, + wantErr: true, + wantErrContains: "invalid allow filter", + }, + { + name: "valid-deny", + defaultFormat: "cloudevents-json", + composedOf: ComposedOfEventArgs{Format: "cloudevents-json", Deny: []string{`"/Data/Header/status" == 401`}}, + wantFlags: &EventFlags{ + Format: "cloudevents-json", + DenyFilters: []string{`"/Data/Header/status" == 401`}, + }, + }, + { + name: "invalid-deny", + defaultFormat: "cloudevents-json", + composedOf: ComposedOfEventArgs{Format: "cloudevents-json", Deny: []string{`"/Data/Header/status" $$== 401`}}, + wantErr: true, + wantErrContains: "invalid deny filter", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + got, err := NewEventFlags(tt.defaultFormat, tt.composedOf) + if tt.wantErr { + require.Error(err) + assert.Nil(got) + if tt.wantErrIs != nil { + assert.ErrorIs(err, tt.wantErrIs) + } + if tt.wantErrContains != "" { + assert.Contains(err.Error(), tt.wantErrContains) + } + return + } + require.NoError(err) + assert.NotNil(got) + assert.Equal(tt.wantFlags, got) + }) + } +} diff --git a/internal/cmd/base/option.go b/internal/cmd/base/option.go index 748ba3d5c..d83119f53 100644 --- a/internal/cmd/base/option.go +++ b/internal/cmd/base/option.go @@ -127,7 +127,8 @@ func WithEventerConfig(config *event.EventerConfig) Option { } } -// WithEventer allows an optional event configuration flags +// WithEventer allows an optional event configuration flags which override +// whatever is in the EventerConfig func WithEventFlags(flags *EventFlags) Option { return func(o *Options) { o.withEventFlags = flags diff --git a/internal/cmd/base/option_test.go b/internal/cmd/base/option_test.go index 269e83968..500ab4a61 100644 --- a/internal/cmd/base/option_test.go +++ b/internal/cmd/base/option_test.go @@ -23,6 +23,7 @@ func Test_GetOpts(t *testing.T) { Format: event.JSONSinkFormat, AuditEnabled: &isTrue, ObservationsEnabled: &isTrue, + SysEventsEnabled: &isTrue, } opts := getOpts(WithEventFlags(&f)) testOpts := getDefaultOptions() diff --git a/internal/cmd/base/server_test.go b/internal/cmd/base/server_test.go index ac641b5f4..6bc912072 100644 --- a/internal/cmd/base/server_test.go +++ b/internal/cmd/base/server_test.go @@ -78,11 +78,13 @@ func TestServer_SetupEventing(t *testing.T) { Format: event.JSONSinkFormat, AuditEnabled: &setTrue, ObservationsEnabled: &setFalse, + SysEventsEnabled: &setFalse, })}, want: func() event.EventerConfig { c := event.DefaultEventerConfig() c.AuditEnabled = true c.ObservationsEnabled = false + c.SysEventsEnabled = false return *c }(), }, @@ -137,7 +139,7 @@ func TestServer_SetupEventing(t *testing.T) { assert, require := assert.New(t), require.New(t) event.TestResetSystEventer(t) - err := tt.s.SetupEventing(tt.logger, tt.lock, tt.opt...) + err := tt.s.SetupEventing(tt.logger, tt.lock, tt.name, tt.opt...) if tt.wantErrMatch != nil || tt.wantErrIs != nil { require.Error(err) assert.Nil(tt.s.Eventer) @@ -164,7 +166,7 @@ func TestServer_AddEventerToContext(t *testing.T) { testLogger := hclog.New(&hclog.LoggerOptions{ Mutex: testLock, }) - testEventer, err := event.NewEventer(testLogger, testLock, event.EventerConfig{}) + testEventer, err := event.NewEventer(testLogger, testLock, "TestServer_AddEventerToContext", event.EventerConfig{}) require.NoError(t, err) tests := []struct { name string diff --git a/internal/cmd/base/servers.go b/internal/cmd/base/servers.go index 94a36a2ec..8bb239c36 100644 --- a/internal/cmd/base/servers.go +++ b/internal/cmd/base/servers.go @@ -141,7 +141,7 @@ func NewServer(cmd *Command) *Server { // SetupEventing will setup the server's eventer and initialize the "system // wide" eventer with a pointer to the same eventer -func (b *Server) SetupEventing(logger hclog.Logger, serializationLock *sync.Mutex, opt ...Option) error { +func (b *Server) SetupEventing(logger hclog.Logger, serializationLock *sync.Mutex, serverName string, opt ...Option) error { const op = "base.(Server).SetupEventing" if logger == nil { @@ -150,7 +150,9 @@ func (b *Server) SetupEventing(logger hclog.Logger, serializationLock *sync.Mute if serializationLock == nil { return berrors.New(berrors.InvalidParameter, op, "missing serialization lock") } - + if serverName == "" { + return berrors.New(berrors.InvalidParameter, op, "missing server name") + } opts := getOpts(opt...) if opts.withEventerConfig != nil { if err := opts.withEventerConfig.Validate(); err != nil { @@ -166,8 +168,8 @@ func (b *Server) SetupEventing(logger hclog.Logger, serializationLock *sync.Mute return berrors.Wrap(err, op, berrors.WithMsg("invalid event flags")) } if opts.withEventFlags.Format != "" { - for _, s := range opts.withEventerConfig.Sinks { - s.Format = opts.withEventFlags.Format + for i := 0; i < len(opts.withEventerConfig.Sinks); i++ { + opts.withEventerConfig.Sinks[i].Format = opts.withEventFlags.Format } } if opts.withEventFlags.AuditEnabled != nil { @@ -176,15 +178,28 @@ func (b *Server) SetupEventing(logger hclog.Logger, serializationLock *sync.Mute if opts.withEventFlags.ObservationsEnabled != nil { opts.withEventerConfig.ObservationsEnabled = *opts.withEventFlags.ObservationsEnabled } + if opts.withEventFlags.SysEventsEnabled != nil { + opts.withEventerConfig.SysEventsEnabled = *opts.withEventFlags.SysEventsEnabled + } + if len(opts.withEventFlags.AllowFilters) > 0 { + for i := 0; i < len(opts.withEventerConfig.Sinks); i++ { + opts.withEventerConfig.Sinks[i].AllowFilters = opts.withEventFlags.AllowFilters + } + } + if len(opts.withEventFlags.DenyFilters) > 0 { + for i := 0; i < len(opts.withEventerConfig.Sinks); i++ { + opts.withEventerConfig.Sinks[i].DenyFilters = opts.withEventFlags.DenyFilters + } + } } - e, err := event.NewEventer(logger, serializationLock, *opts.withEventerConfig) + e, err := event.NewEventer(logger, serializationLock, serverName, *opts.withEventerConfig) if err != nil { return berrors.Wrap(err, op, berrors.WithMsg("unable to create eventer")) } b.Eventer = e - if err := event.InitSysEventer(logger, serializationLock, event.WithEventer(e)); err != nil { + if err := event.InitSysEventer(logger, serializationLock, serverName, event.WithEventer(e)); err != nil { return berrors.Wrap(err, op, berrors.WithMsg("unable to initialize system eventer")) } diff --git a/internal/cmd/commands/database/init.go b/internal/cmd/commands/database/init.go index e56f2b5d7..ffe1ce407 100644 --- a/internal/cmd/commands/database/init.go +++ b/internal/cmd/commands/database/init.go @@ -185,7 +185,18 @@ func (c *InitCommand) Run(args []string) (retCode int) { return base.CommandCliError } - if err := c.srv.SetupEventing(c.srv.Logger, c.srv.StderrLock, base.WithEventerConfig(c.Config.Eventing)); err != nil { + var serverName string + switch { + case c.Config.Controller == nil: + serverName = "boundary-database-init" + default: + if _, err := c.Config.Controller.InitNameIfEmpty(); err != nil { + c.UI.Error(err.Error()) + return base.CommandCliError + } + serverName = c.Config.Controller.Name + "/boundary-database-init" + } + if err := c.srv.SetupEventing(c.srv.Logger, c.srv.StderrLock, serverName, base.WithEventerConfig(c.Config.Eventing)); err != nil { c.UI.Error(err.Error()) return base.CommandCliError } diff --git a/internal/cmd/commands/database/migrate.go b/internal/cmd/commands/database/migrate.go index 22c7338ce..514c453c3 100644 --- a/internal/cmd/commands/database/migrate.go +++ b/internal/cmd/commands/database/migrate.go @@ -136,8 +136,18 @@ func (c *MigrateCommand) Run(args []string) (retCode int) { c.UI.Error(err.Error()) return base.CommandCliError } - - if err := c.srv.SetupEventing(c.srv.Logger, c.srv.StderrLock, base.WithEventerConfig(c.Config.Eventing)); err != nil { + var serverName string + switch { + case c.Config.Controller == nil: + serverName = "boundary-database-migrate" + default: + if _, err := c.Config.Controller.InitNameIfEmpty(); err != nil { + c.UI.Error(err.Error()) + return base.CommandCliError + } + serverName = c.Config.Controller.Name + "/boundary-database-migrate" + } + if err := c.srv.SetupEventing(c.srv.Logger, c.srv.StderrLock, serverName, base.WithEventerConfig(c.Config.Eventing)); err != nil { c.UI.Error(err.Error()) return base.CommandCliError } diff --git a/internal/cmd/commands/dev/dev.go b/internal/cmd/commands/dev/dev.go index 5ba25f908..fb0b229ab 100644 --- a/internal/cmd/commands/dev/dev.go +++ b/internal/cmd/commands/dev/dev.go @@ -14,6 +14,7 @@ import ( "github.com/hashicorp/boundary/internal/host/static" "github.com/hashicorp/boundary/internal/iam" "github.com/hashicorp/boundary/internal/intglobals" + "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/boundary/internal/servers/controller" "github.com/hashicorp/boundary/internal/servers/controller/handlers" "github.com/hashicorp/boundary/internal/servers/worker" @@ -66,6 +67,12 @@ type Command struct { flagDatabaseUrl string flagContainerImage string flagDisableDatabaseDestruction bool + flagEventFormat string + flagAudit string + flagObservations string + flagSysEvents string + flagEveryEventAllowFilters []string + flagEveryEventDenyFilters []string } func (c *Command) Synopsis() string { @@ -262,6 +269,40 @@ func (c *Command) Flags() *base.FlagSets { Target: &c.flagContainerImage, Usage: `Specifies a container image to be utilized. Must be in : format`, }) + f.StringVar(&base.StringVar{ + Name: "event-format", + Target: &c.flagEventFormat, + Completion: complete.PredictSet("cloudevents-json", "cloudevents-text"), + Usage: `Event format. Supported values are "cloudevents-json" and "cloudevents-text".`, + }) + f.StringVar(&base.StringVar{ + Name: "observation-events", + Target: &c.flagObservations, + Completion: complete.PredictSet("true", "false"), + Usage: `Emit observation events. Supported values are "true" and "false".`, + }) + f.StringVar(&base.StringVar{ + Name: "audit-events", + Target: &c.flagAudit, + Completion: complete.PredictSet("true", "false"), + Usage: `Emit audit events. Supported values are "true" and "false".`, + }) + f.StringVar(&base.StringVar{ + Name: "system-events", + Target: &c.flagSysEvents, + Completion: complete.PredictSet("true", "false"), + Usage: `Emit system events. Supported values are "true" and "false".`, + }) + f.StringSliceVar(&base.StringSliceVar{ + Name: "event-allow-filter", + Target: &c.flagEveryEventAllowFilters, + Usage: `The optional every event allow filter. May be specified multiple times.`, + }) + f.StringSliceVar(&base.StringSliceVar{ + Name: "event-deny-filter", + Target: &c.flagEveryEventDenyFilters, + Usage: `The optional every event deny filter. May be specified multiple times.`, + }) return set } @@ -420,7 +461,30 @@ func (c *Command) Run(args []string) int { return base.CommandUserError } - if err := c.SetupEventing(c.Logger, c.StderrLock, base.WithEventerConfig(c.Config.Eventing)); err != nil { + var serverName string + switch { + case c.Config.Controller == nil: + serverName = "boundary-dev" + default: + if _, err := c.Config.Controller.InitNameIfEmpty(); err != nil { + c.UI.Error(err.Error()) + return base.CommandCliError + } + serverName = c.Config.Controller.Name + "/boundary-dev" + } + eventFlags, err := base.NewEventFlags(event.TextSinkFormat, base.ComposedOfEventArgs{ + Format: c.flagEventFormat, + Audit: c.flagAudit, + Observations: c.flagObservations, + SysEvents: c.flagSysEvents, + Allow: c.flagEveryEventAllowFilters, + Deny: c.flagEveryEventDenyFilters, + }) + if err != nil { + c.UI.Error(err.Error()) + return base.CommandUserError + } + if err := c.SetupEventing(c.Logger, c.StderrLock, serverName, base.WithEventerConfig(c.Config.Eventing), base.WithEventFlags(eventFlags)); err != nil { c.UI.Error(err.Error()) return base.CommandCliError } diff --git a/internal/cmd/commands/server/server.go b/internal/cmd/commands/server/server.go index f646e340a..fd4b72ab7 100644 --- a/internal/cmd/commands/server/server.go +++ b/internal/cmd/commands/server/server.go @@ -147,7 +147,23 @@ func (c *Command) Run(args []string) int { return base.CommandUserError } - if err := c.SetupEventing(c.Logger, c.StderrLock, base.WithEventerConfig(c.Config.Eventing)); err != nil { + var serverNames []string + if c.Config.Controller != nil { + if _, err := c.Config.Controller.InitNameIfEmpty(); err != nil { + c.UI.Error(err.Error()) + return base.CommandUserError + } + serverNames = append(serverNames, c.Config.Controller.Name) + } + if c.Config.Worker != nil { + if _, err := c.Config.Worker.InitNameIfEmpty(); err != nil { + c.UI.Error(err.Error()) + return base.CommandUserError + } + serverNames = append(serverNames, c.Config.Worker.Name) + } + + if err := c.SetupEventing(c.Logger, c.StderrLock, strings.Join(serverNames, "/"), base.WithEventerConfig(c.Config.Eventing)); err != nil { c.UI.Error(err.Error()) return base.CommandUserError } diff --git a/internal/cmd/commands/server/server_test.go b/internal/cmd/commands/server/server_test.go index 1a569eafa..5a7db61c6 100644 --- a/internal/cmd/commands/server/server_test.go +++ b/internal/cmd/commands/server/server_test.go @@ -29,7 +29,7 @@ func testServerCommand(t *testing.T, controllerKey string) *Command { } require.NoError(cmd.SetupLogging("trace", "", "", "")) - require.NoError(cmd.SetupEventing(cmd.Logger, cmd.StderrLock)) + require.NoError(cmd.SetupEventing(cmd.Logger, cmd.StderrLock, "test-server-command")) kmsHcl := fmt.Sprintf(rootKmsConfig, controllerKey) parsedKmsConfig, err := config.Parse(kmsHcl) diff --git a/internal/cmd/config/config.go b/internal/cmd/config/config.go index ab9b99fab..f001cd62b 100644 --- a/internal/cmd/config/config.go +++ b/internal/cmd/config/config.go @@ -13,6 +13,7 @@ import ( "github.com/hashicorp/boundary/internal/observability/event" wrapping "github.com/hashicorp/go-kms-wrapping" + "github.com/hashicorp/go-secure-stdlib/base62" "github.com/hashicorp/go-secure-stdlib/configutil" "github.com/hashicorp/go-secure-stdlib/parseutil" "github.com/hashicorp/go-secure-stdlib/strutil" @@ -128,6 +129,16 @@ type Controller struct { StatusGracePeriodDuration time.Duration `hcl:"-"` } +func (c *Controller) InitNameIfEmpty() (string, error) { + if c == nil { + return "", fmt.Errorf("controller config is empty") + } + if err := initNameIfEmpty(&c.Name); err != nil { + return "", fmt.Errorf("error auto-generating controller name: %w", err) + } + return c.Name, nil +} + type Worker struct { Name string `hcl:"name"` Description string `hcl:"description"` @@ -148,6 +159,26 @@ type Worker struct { StatusGracePeriodDuration time.Duration `hcl:"-"` } +func (w *Worker) InitNameIfEmpty() (string, error) { + if w == nil { + return "", fmt.Errorf("worker config is empty") + } + if err := initNameIfEmpty(&w.Name); err != nil { + return "", fmt.Errorf("error auto-generating worker name: %w", err) + } + return w.Name, nil +} + +func initNameIfEmpty(name *string) error { + if *name == "" { + var err error + if *name, err = base62.Random(10); err != nil { + return err + } + } + return nil +} + type Database struct { Url string `hcl:"url"` MigrationUrl string `hcl:"migration_url"` diff --git a/internal/filter/filter.go b/internal/filter/filter.go new file mode 100644 index 000000000..ce8071b8e --- /dev/null +++ b/internal/filter/filter.go @@ -0,0 +1,43 @@ +package filter + +import ( + "reflect" + + "google.golang.org/protobuf/types/known/structpb" + "google.golang.org/protobuf/types/known/timestamppb" + "google.golang.org/protobuf/types/known/wrapperspb" +) + +// WellKnownTypeFilterHook is passed to bexpr to treat all proto well-known +// types as the types they wrap for comparison. +func WellKnownTypeFilterHook(v reflect.Value) reflect.Value { + if !v.CanInterface() { + return v + } + ret := v.Interface() + switch pm := v.Interface().(type) { + case *wrapperspb.BoolValue: + ret = pm.GetValue() + case *wrapperspb.BytesValue: + ret = pm.GetValue() + case *wrapperspb.StringValue: + ret = pm.GetValue() + case *wrapperspb.DoubleValue: + ret = pm.GetValue() + case *wrapperspb.FloatValue: + ret = pm.GetValue() + case *wrapperspb.Int32Value: + ret = pm.GetValue() + case *wrapperspb.Int64Value: + ret = pm.GetValue() + case *wrapperspb.UInt32Value: + ret = pm.GetValue() + case *wrapperspb.UInt64Value: + ret = pm.GetValue() + case *structpb.Struct: + ret = pm.AsMap() + case *timestamppb.Timestamp: + ret = pm.AsTime() + } + return reflect.ValueOf(ret) +} diff --git a/internal/filter/filter_test.go b/internal/filter/filter_test.go new file mode 100644 index 000000000..aacc5613d --- /dev/null +++ b/internal/filter/filter_test.go @@ -0,0 +1,53 @@ +package filter + +import ( + "reflect" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/structpb" + "google.golang.org/protobuf/types/known/timestamppb" + "google.golang.org/protobuf/types/known/wrapperspb" +) + +func TestWellKnownTypeFilterHook(t *testing.T) { + t.Run("wrappers", func(t *testing.T) { + conversions := map[interface{}]interface{}{ + wrapperspb.String("foo"): "foo", + wrapperspb.UInt64(123): uint64(123), + wrapperspb.Int64(123): int64(123), + wrapperspb.UInt32(123): uint32(123), + wrapperspb.Int32(123): int32(123), + wrapperspb.Float(123): float32(123), + wrapperspb.Double(123): float64(123), + wrapperspb.Bytes([]byte("foo")): []byte("foo"), + wrapperspb.Bool(true): true, + } + for in, out := range conversions { + assert.Equal(t, reflect.ValueOf(out).Interface(), WellKnownTypeFilterHook(reflect.ValueOf(in)).Interface()) + } + }) + t.Run("time", func(t *testing.T) { + assert := assert.New(t) + now := time.Now() + ts := timestamppb.New(now) + expect := reflect.ValueOf(ts.AsTime()) + actual := WellKnownTypeFilterHook(reflect.ValueOf(ts)) + assert.Equal(expect.Interface(), actual.Interface()) + }) + t.Run("struct", func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + s, err := structpb.NewStruct(map[string]interface{}{ + "name": "test", + }) + require.NoError(err) + + expect := map[string]interface{}{ + "name": "test", + } + actual := WellKnownTypeFilterHook(reflect.ValueOf(s)) + assert.Equal(expect, actual.Interface()) + }) +} diff --git a/internal/observability/event/context.go b/internal/observability/event/context.go index b2969f143..92bd6c51a 100644 --- a/internal/observability/event/context.go +++ b/internal/observability/event/context.go @@ -52,6 +52,9 @@ func NewRequestInfoContext(ctx context.Context, info *RequestInfo) (context.Cont if info.Id == "" { return nil, fmt.Errorf("%s: missing request info id: %w", op, ErrInvalidParameter) } + if info.EventId == "" { + return nil, fmt.Errorf("%s: missing request info event id: %w", op, ErrInvalidParameter) + } return context.WithValue(ctx, requestInfoKey, info), nil } @@ -211,9 +214,9 @@ func addCtxOptions(ctx context.Context, opt ...Option) ([]Option, error) { reqInfo, ok := RequestInfoFromContext(ctx) if !ok { // there's no RequestInfo, so there's no id associated with the - // observation and we'll generate one and flush the observation + // event and we'll generate one and flush the event // since there will never be another with the same id - id, err := newId(string(ObservationType)) + id, err := NewId(idPrefix) if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } @@ -224,15 +227,12 @@ func addCtxOptions(ctx context.Context, opt ...Option) ([]Option, error) { return retOpts, nil } retOpts = append(retOpts, WithRequestInfo(reqInfo)) - if reqInfo.Id != "" { - retOpts = append(retOpts, WithId(reqInfo.Id)) - } - switch reqInfo.Id { + switch reqInfo.EventId { case "": - // there's no RequestInfo.Id associated with the observation, so we'll - // generate one and flush the observation since there will never be - // another with the same id - id, err := newId(string(ObservationType)) + // there's no RequestInfo.EventId associated with the observation, + // so we'll generate one and flush the observation since there will + // never be another with the same id + id, err := NewId("e") if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } @@ -242,7 +242,7 @@ func addCtxOptions(ctx context.Context, opt ...Option) ([]Option, error) { } return retOpts, nil default: - retOpts = append(retOpts, WithId(reqInfo.Id)) + retOpts = append(retOpts, WithId(reqInfo.EventId)) } } return retOpts, nil @@ -274,7 +274,7 @@ func WriteSysEvent(ctx context.Context, caller Op, data map[string]interface{}) return } - id, err := newId(string(SystemType)) + id, err := NewId(string(SystemType)) if err != nil { eventer.logger.Error(fmt.Sprintf("%s: %v", op, err)) eventer.logger.Error(fmt.Sprintf("%s: unable to generate id while writing sysevent: (%s) %+v", op, caller, data)) diff --git a/internal/observability/event/context_test.go b/internal/observability/event/context_test.go index a3804c48a..5e44beb7d 100644 --- a/internal/observability/event/context_test.go +++ b/internal/observability/event/context_test.go @@ -17,6 +17,7 @@ import ( "github.com/hashicorp/boundary/internal/host/static" "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/boundary/internal/types/scope" + "github.com/hashicorp/eventlogger/formatter_filters/cloudevents" "github.com/hashicorp/go-hclog" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -48,6 +49,9 @@ func Test_NewRequestInfoContext(t *testing.T) { testInfoMissingId := event.TestRequestInfo(t) testInfoMissingId.Id = "" + testInfoMissingEventId := event.TestRequestInfo(t) + testInfoMissingEventId.EventId = "" + tests := []struct { name string ctx context.Context @@ -74,6 +78,13 @@ func Test_NewRequestInfoContext(t *testing.T) { wantErrIs: event.ErrInvalidParameter, wantErrContains: "missing request info id", }, + { + name: "missing-request-info-event-id", + ctx: context.Background(), + requestInfo: testInfoMissingEventId, + wantErrIs: event.ErrInvalidParameter, + wantErrContains: "missing request info event id", + }, { name: "valid", ctx: context.Background(), @@ -152,7 +163,7 @@ func Test_NewEventerContext(t *testing.T) { Mutex: testLock, Name: "test", }) - testEventer, err := event.NewEventer(testLogger, testLock, testSetup.EventerConfig) + testEventer, err := event.NewEventer(testLogger, testLock, "Test_NewEventerContext", testSetup.EventerConfig) require.NoError(t, err) tests := []struct { name string @@ -209,7 +220,7 @@ func Test_EventerFromContext(t *testing.T) { Mutex: testLock, Name: "test", }) - testEventer, err := event.NewEventer(testLogger, testLock, testSetup.EventerConfig) + testEventer, err := event.NewEventer(testLogger, testLock, "Test_EventerFromContext", testSetup.EventerConfig) require.NoError(t, err) testEventerCtx, err := event.NewEventerContext(context.Background(), testEventer) @@ -265,22 +276,23 @@ func Test_WriteObservation(t *testing.T) { Mutex: testLock, Name: "test", }) - e, err := event.NewEventer(testLogger, testLock, c.EventerConfig) + e, err := event.NewEventer(testLogger, testLock, "Test_WriteObservation", c.EventerConfig) require.NoError(t, err) - info := &event.RequestInfo{Id: "867-5309"} + info := &event.RequestInfo{Id: "867-5309", EventId: "411"} testCtx, err := event.NewEventerContext(context.Background(), e) require.NoError(t, err) testCtx, err = event.NewRequestInfoContext(testCtx, info) require.NoError(t, err) - testCtxNoInfoId, err := event.NewEventerContext(context.Background(), e) + testCtxNoEventInfoId, err := event.NewEventerContext(context.Background(), e) require.NoError(t, err) - noId := &event.RequestInfo{Id: "867-5309"} - testCtxNoInfoId, err = event.NewRequestInfoContext(testCtxNoInfoId, noId) + noEventId := &event.RequestInfo{Id: "867-5309", EventId: "411"} + testCtxNoEventInfoId, err = event.NewRequestInfoContext(testCtxNoEventInfoId, noEventId) require.NoError(t, err) - noId.Id = "" + noEventId.EventId = "" + noEventId.Id = "" type observationPayload struct { header map[string]interface{} @@ -329,9 +341,9 @@ func Test_WriteObservation(t *testing.T) { wantErrContains string }{ { - name: "no-info-id", + name: "no-info-event-id", noFlush: true, - ctx: testCtxNoInfoId, + ctx: testCtxNoEventInfoId, observationPayload: []observationPayload{ { header: map[string]interface{}{ @@ -344,7 +356,7 @@ func Test_WriteObservation(t *testing.T) { }, observationSinkFileName: c.AllEvents.Name(), setup: func() error { - return event.InitSysEventer(testLogger, testLock, event.WithEventerConfig(&c.EventerConfig)) + return event.InitSysEventer(testLogger, testLock, "no-info-event-id", event.WithEventerConfig(&c.EventerConfig)) }, cleanup: func() { event.TestResetSystEventer(t) }, }, @@ -395,7 +407,7 @@ func Test_WriteObservation(t *testing.T) { }, observationSinkFileName: c.AllEvents.Name(), setup: func() error { - return event.InitSysEventer(testLogger, testLock, event.WithEventerConfig(&c.EventerConfig)) + return event.InitSysEventer(testLogger, testLock, "use-syseventer", event.WithEventerConfig(&c.EventerConfig)) }, cleanup: func() { event.TestResetSystEventer(t) }, }, @@ -442,12 +454,13 @@ func Test_WriteObservation(t *testing.T) { b, err := ioutil.ReadFile(tt.observationSinkFileName) assert.NoError(err) - gotObservation := &eventJson{} + gotObservation := &cloudevents.Event{} err = json.Unmarshal(b, gotObservation) require.NoErrorf(err, "json: %s", string(b)) actualJson, err := json.Marshal(gotObservation) require.NoError(err) + fmt.Println(string(actualJson)) wantJson := testObservationJsonFromCtx(t, tt.ctx, event.Op(tt.name), gotObservation, tt.header, tt.details) assert.JSONEq(string(wantJson), string(actualJson)) @@ -464,7 +477,7 @@ func Test_WriteObservation(t *testing.T) { Mutex: testLock, Name: "test", }) - e, err := event.NewEventer(testLogger, testLock, c.EventerConfig) + e, err := event.NewEventer(testLogger, testLock, "not-enabled", c.EventerConfig) require.NoError(err) testCtx, err := event.NewEventerContext(context.Background(), e) @@ -481,36 +494,127 @@ func Test_WriteObservation(t *testing.T) { assert.NoError(err) assert.Len(b, 0) }) + } -func testObservationJsonFromCtx(t *testing.T, ctx context.Context, caller event.Op, got *eventJson, hdr, details map[string]interface{}) []byte { +func Test_Filtering(t *testing.T) { + event.TestEnableEventing(t, true) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + tests := []struct { + name string + allow []string + deny []string + hdr map[string]interface{} + found bool + }{ + { + name: "allowed", + allow: []string{`"/Data/list" contains "1"`}, + hdr: map[string]interface{}{ + "list": []string{"1", "2"}, + }, + found: true, + }, + { + name: "not-allowed", + allow: []string{`"/Data/list" contains "22"`}, + hdr: map[string]interface{}{ + "list": []string{"1", "2"}, + }, + found: false, + }, + { + name: "deny", + deny: []string{`"/Data/list" contains "1"`}, + hdr: map[string]interface{}{ + "list": []string{"1", "2"}, + }, + found: false, + }, + { + name: "not-deny", + deny: []string{`"/Data/list" contains "22"`}, + hdr: map[string]interface{}{ + "list": []string{"1", "2"}, + }, + found: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + info := &event.RequestInfo{Id: "867-5309", EventId: "411"} + + c := event.TestEventerConfig(t, "WriteObservation-filtering") + c.EventerConfig.Sinks[0].AllowFilters = tt.allow + c.EventerConfig.Sinks[0].DenyFilters = tt.deny + + e, err := event.NewEventer(testLogger, testLock, "filtering", c.EventerConfig) + require.NoError(err) + + testCtx, err := event.NewEventerContext(context.Background(), e) + require.NoError(err) + testCtx, err = event.NewRequestInfoContext(testCtx, info) + require.NoError(err) + + require.NoError(event.WriteObservation(testCtx, "not-enabled", event.WithHeader(tt.hdr), event.WithFlush())) + + b, err := ioutil.ReadFile(c.AllEvents.Name()) + assert.NoError(err) + switch tt.found { + case true: + assert.NotEmpty(b) + case false: + assert.Empty(b) + } + }) + } +} + +func Test_DefaultEventerConfig(t *testing.T) { + t.Run("assert-default", func(t *testing.T) { + assert.Equal(t, &event.EventerConfig{ + AuditEnabled: false, + ObservationsEnabled: true, + SysEventsEnabled: true, + Sinks: []event.SinkConfig{event.DefaultSink()}, + }, event.DefaultEventerConfig()) + }) +} + +func testObservationJsonFromCtx(t *testing.T, ctx context.Context, caller event.Op, got *cloudevents.Event, hdr, details map[string]interface{}) []byte { t.Helper() require := require.New(t) reqInfo, _ := event.RequestInfoFromContext(ctx) // require.Truef(ok, "missing reqInfo in ctx") - j := eventJson{ - CreatedAt: got.CreatedAt, - EventType: string(event.ObservationType), - Payload: map[string]interface{}{ - event.IdField: got.Payload[event.IdField].(string), - event.HeaderField: map[string]interface{}{ - event.RequestInfoField: reqInfo, - event.VersionField: testObservationVersion, - }, + j := cloudevents.Event{ + ID: got.ID, + Time: got.Time, + Source: got.Source, + SpecVersion: got.SpecVersion, + Type: got.Type, + DataContentType: got.DataContentType, + Data: map[string]interface{}{ + event.RequestInfoField: reqInfo, + event.VersionField: testObservationVersion, }, } if hdr != nil { - h := j.Payload[event.HeaderField].(map[string]interface{}) + h := j.Data.(map[string]interface{}) for k, v := range hdr { h[k] = v } } if details != nil { details[event.OpField] = string(caller) - d := got.Payload[event.DetailsField].([]interface{})[0].(map[string]interface{}) - j.Payload[event.DetailsField] = []struct { + d := got.Data.(map[string]interface{})[event.DetailsField].([]interface{})[0].(map[string]interface{}) + j.Data.(map[string]interface{})[event.DetailsField] = []struct { CreatedAt string `json:"created_at"` Type string `json:"type"` Payload map[string]interface{} `json:"payload"` @@ -527,12 +631,6 @@ func testObservationJsonFromCtx(t *testing.T, ctx context.Context, caller event. return b } -type eventJson struct { - CreatedAt string `json:"created_at"` - EventType string `json:"event_type"` - Payload map[string]interface{} `json:"payload"` -} - func Test_WriteAudit(t *testing.T) { event.TestEnableEventing(t, true) @@ -546,10 +644,10 @@ func Test_WriteAudit(t *testing.T) { Mutex: testLock, Name: "test", }) - e, err := event.NewEventer(testLogger, testLock, c.EventerConfig) + e, err := event.NewEventer(testLogger, testLock, "Test_WriteAudit", c.EventerConfig) require.NoError(t, err) - info := &event.RequestInfo{Id: "867-5309"} + info := &event.RequestInfo{Id: "867-5309", EventId: "411"} ctx, err := event.NewEventerContext(context.Background(), e) require.NoError(t, err) @@ -653,7 +751,7 @@ func Test_WriteAudit(t *testing.T) { Request: testReq, }, setup: func() error { - return event.InitSysEventer(testLogger, testLock, event.WithEventerConfig(&c.EventerConfig)) + return event.InitSysEventer(testLogger, testLock, "use-syseventer", event.WithEventerConfig(&c.EventerConfig)) }, cleanup: func() { event.TestResetSystEventer(t) }, auditSinkFileName: c.AllEvents.Name(), @@ -671,7 +769,7 @@ func Test_WriteAudit(t *testing.T) { }, }, wantAudit: &testAudit{ - Id: "867-5309", + Id: "411", Auth: testAuth, Request: testReq, Response: testResp, @@ -713,19 +811,22 @@ func Test_WriteAudit(t *testing.T) { b, err := ioutil.ReadFile(tt.auditSinkFileName) require.NoError(err) - gotAudit := &eventJson{} + gotAudit := &cloudevents.Event{} err = json.Unmarshal(b, gotAudit) require.NoErrorf(err, "json: %s", string(b)) actualJson, err := json.Marshal(gotAudit) require.NoError(err) - - wantEvent := eventJson{ - CreatedAt: gotAudit.CreatedAt, - EventType: string(gotAudit.EventType), - Payload: map[string]interface{}{ + wantEvent := cloudevents.Event{ + ID: gotAudit.ID, + Source: gotAudit.Source, + SpecVersion: gotAudit.SpecVersion, + DataContentType: gotAudit.DataContentType, + Time: gotAudit.Time, + Type: "audit", + Data: map[string]interface{}{ "auth": tt.wantAudit.Auth, - "id": gotAudit.Payload["id"], + "id": gotAudit.Data.(map[string]interface{})["id"], "timestamp": now, "request": tt.wantAudit.Request, "serialized_hmac": "", @@ -734,13 +835,13 @@ func Test_WriteAudit(t *testing.T) { }, } if tt.wantAudit.Id != "" { - wantEvent.Payload["id"] = tt.wantAudit.Id - wantEvent.Payload["request_info"] = event.RequestInfo{ - Id: tt.wantAudit.Id, + wantEvent.Data.(map[string]interface{})["id"] = tt.wantAudit.Id + wantEvent.Data.(map[string]interface{})["request_info"] = event.RequestInfo{ + Id: gotAudit.Data.(map[string]interface{})["request_info"].(map[string]interface{})["id"].(string), } } if tt.wantAudit.Response != nil { - wantEvent.Payload["response"] = tt.wantAudit.Response + wantEvent.Data.(map[string]interface{})["response"] = tt.wantAudit.Response } wantJson, err := json.Marshal(wantEvent) require.NoError(err) @@ -758,7 +859,7 @@ func Test_WriteAudit(t *testing.T) { Mutex: testLock, Name: "test", }) - e, err := event.NewEventer(testLogger, testLock, c.EventerConfig) + e, err := event.NewEventer(testLogger, testLock, "not-enabled", c.EventerConfig) require.NoError(err) testCtx, err := event.NewEventerContext(context.Background(), e) @@ -786,10 +887,10 @@ func Test_WriteError(t *testing.T) { Mutex: testLock, Name: "test", }) - e, err := event.NewEventer(testLogger, testLock, c.EventerConfig, event.WithNow(now)) + e, err := event.NewEventer(testLogger, testLock, "Test_WriteError", c.EventerConfig, event.WithNow(now)) require.NoError(t, err) - info := &event.RequestInfo{Id: "867-5309"} + info := &event.RequestInfo{Id: "867-5309", EventId: "411"} testCtx, err := event.NewEventerContext(context.Background(), e) require.NoError(t, err) @@ -798,10 +899,11 @@ func Test_WriteError(t *testing.T) { testCtxNoInfoId, err := event.NewEventerContext(context.Background(), e) require.NoError(t, err) - noId := &event.RequestInfo{Id: "867-5309"} + noId := &event.RequestInfo{Id: "867-5309", EventId: "411"} testCtxNoInfoId, err = event.NewRequestInfoContext(testCtxNoInfoId, noId) require.NoError(t, err) noId.Id = "" + noId.EventId = "" testError := fakeError{ Msg: "test", @@ -838,7 +940,7 @@ func Test_WriteError(t *testing.T) { ctx: context.Background(), e: &testError, setup: func() error { - return event.InitSysEventer(testLogger, testLock, event.WithEventerConfig(&c.EventerConfig)) + return event.InitSysEventer(testLogger, testLock, "use-syseventer", event.WithEventerConfig(&c.EventerConfig)) }, cleanup: func() { event.TestResetSystEventer(t) }, errSinkFileName: c.ErrorEvents.Name(), @@ -849,7 +951,7 @@ func Test_WriteError(t *testing.T) { e: &testError, info: &event.RequestInfo{}, setup: func() error { - return event.InitSysEventer(testLogger, testLock, event.WithEventerConfig(&c.EventerConfig)) + return event.InitSysEventer(testLogger, testLock, "no-info-id", event.WithEventerConfig(&c.EventerConfig)) }, cleanup: func() { event.TestResetSystEventer(t) }, errSinkFileName: c.ErrorEvents.Name(), @@ -887,15 +989,15 @@ func Test_WriteError(t *testing.T) { return } - gotError := &eventJson{} + gotError := &cloudevents.Event{} err = json.Unmarshal(b, gotError) require.NoErrorf(err, "json: %s", string(b)) require.NoError(err) actualError := fakeError{ - Msg: gotError.Payload["error"].(map[string]interface{})["Msg"].(string), - Code: gotError.Payload["error"].(map[string]interface{})["Code"].(string), + Msg: gotError.Data.(map[string]interface{})["error"].(map[string]interface{})["Msg"].(string), + Code: gotError.Data.(map[string]interface{})["error"].(map[string]interface{})["Code"].(string), } assert.Equal(tt.e, &actualError) @@ -959,7 +1061,7 @@ func Test_WriteSysEvent(t *testing.T) { ctx: context.Background(), data: map[string]interface{}{"data": "test-data", event.ServerName: "test-server", event.ServerAddress: "localhost"}, setup: func() error { - return event.InitSysEventer(testLogger, testLock, event.WithEventerConfig(&c.EventerConfig)) + return event.InitSysEventer(testLogger, testLock, "use-syseventer", event.WithEventerConfig(&c.EventerConfig)) }, cleanup: func() { event.TestResetSystEventer(t) }, sinkFileName: c.AllEvents.Name(), @@ -990,11 +1092,11 @@ func Test_WriteSysEvent(t *testing.T) { return } - gotSysEvent := &eventJson{} + gotSysEvent := &cloudevents.Event{} err = json.Unmarshal(b, gotSysEvent) require.NoErrorf(err, "json: %s", string(b)) - assert.Equal(tt.data, gotSysEvent.Payload["data"].(map[string]interface{})) + assert.Equal(tt.data, gotSysEvent.Data.(map[string]interface{})["data"].(map[string]interface{})) } }) } diff --git a/internal/observability/event/event.go b/internal/observability/event/event.go index 8443e0ed8..6d935bba5 100644 --- a/internal/observability/event/event.go +++ b/internal/observability/event/event.go @@ -1,6 +1,8 @@ package event -import "google.golang.org/protobuf/proto" +import ( + "google.golang.org/protobuf/proto" +) type ( Id string @@ -9,6 +11,7 @@ type ( // RequestInfo defines the fields captured about a Boundary request. type RequestInfo struct { + EventId string `json:"-"` Id string `json:"id,omitempty"` Method string `json:"method,omitempty"` Path string `json:"path,omitempty"` diff --git a/internal/observability/event/event_audit.go b/internal/observability/event/event_audit.go index 1c805d5f2..1175bb657 100644 --- a/internal/observability/event/event_audit.go +++ b/internal/observability/event/event_audit.go @@ -39,7 +39,7 @@ func newAudit(fromOperation Op, opt ...Option) (*audit, error) { opts := getOpts(opt...) if opts.withId == "" { var err error - opts.withId, err = newId(string(AuditType)) + opts.withId, err = NewId(string(AuditType)) if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } diff --git a/internal/observability/event/event_error.go b/internal/observability/event/event_error.go index 8a358794a..ca537c498 100644 --- a/internal/observability/event/event_error.go +++ b/internal/observability/event/event_error.go @@ -26,7 +26,7 @@ func newError(fromOperation Op, e error, opt ...Option) (*err, error) { opts := getOpts(opt...) if opts.withId == "" { var err error - opts.withId, err = newId(string(ErrorType)) + opts.withId, err = NewId(string(ErrorType)) if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } diff --git a/internal/observability/event/event_observation.go b/internal/observability/event/event_observation.go index 16849f83c..1ee91041e 100644 --- a/internal/observability/event/event_observation.go +++ b/internal/observability/event/event_observation.go @@ -3,6 +3,7 @@ package event import ( "fmt" + "github.com/hashicorp/eventlogger" "github.com/hashicorp/eventlogger/filters/gated" "github.com/hashicorp/go-secure-stdlib/strutil" ) @@ -11,10 +12,13 @@ import ( const observationVersion = "v0.1" type observation struct { - *gated.Payload - Version string `json:"version"` - Op Op `json:"op,omitempty"` - RequestInfo *RequestInfo `json:"request_info,omitempty"` + Version string `json:"version"` + Op Op `json:"op,omitempty"` + RequestInfo *RequestInfo `json:"request_info,omitempty"` + ID string `json:"-"` + Flush bool `json:"-"` + Header map[string]interface{} `json:"header,omitempty"` + Detail map[string]interface{} `json:"detail,omitempty"` } func newObservation(fromOperation Op, opt ...Option) (*observation, error) { @@ -25,7 +29,7 @@ func newObservation(fromOperation Op, opt ...Option) (*observation, error) { opts := getOpts(opt...) if opts.withId == "" { var err error - opts.withId, err = newId(string(ObservationType)) + opts.withId, err = NewId(string(ObservationType)) if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } @@ -36,12 +40,10 @@ func newObservation(fromOperation Op, opt ...Option) (*observation, error) { } } i := &observation{ - Payload: &gated.Payload{ - ID: opts.withId, - Header: opts.withHeader, - Detail: opts.withDetails, - Flush: opts.withFlush, - }, + ID: opts.withId, + Header: opts.withHeader, + Detail: opts.withDetails, + Flush: opts.withFlush, Op: fromOperation, RequestInfo: opts.withRequestInfo, Version: observationVersion, @@ -65,3 +67,51 @@ func (i *observation) validate() error { } return nil } + +// ComposedFrom will build a single event payload which will be +// Flushed/Processed from a collection of gated observation events. The payload +// returned is not a Gateable payload intentionally. Note: the receiver is +// always nil when this function is called. +func (o *observation) ComposeFrom(events []*eventlogger.Event) (eventlogger.EventType, interface{}, error) { + const op = "event.(observation).ComposedFrom" + if len(events) == 0 { + return "", nil, fmt.Errorf("%s: missing events: %w", op, eventlogger.ErrInvalidParameter) + } + + payload := map[string]interface{}{} + for i, v := range events { + g, ok := v.Payload.(*observation) + if !ok { + return "", nil, fmt.Errorf("%s: event %d is not an observation: %w", op, i, eventlogger.ErrInvalidParameter) + } + if g.Header != nil { + for hdrK, hdrV := range g.Header { + payload[hdrK] = hdrV + } + } + if g.Detail != nil { + if _, ok := payload[DetailsField]; !ok { + payload[DetailsField] = []gated.EventPayloadDetails{} + } + payload[DetailsField] = append(payload[DetailsField].([]gated.EventPayloadDetails), gated.EventPayloadDetails{ + Type: string(v.Type), + CreatedAt: v.CreatedAt.String(), + Payload: g.Detail, + }) + } + } + return events[0].Type, payload, nil +} + +var _ gated.Gateable = &observation{} + +// GetID returns the unique id used for gating +func (o *observation) GetID() string { + return o.ID +} + +// FlushEvent tells the Filter to flush/process the events associated with +// the Gateable ID +func (o *observation) FlushEvent() bool { + return o.Flush +} diff --git a/internal/observability/event/event_observation_test.go b/internal/observability/event/event_observation_test.go index 73173ef82..8744670b9 100644 --- a/internal/observability/event/event_observation_test.go +++ b/internal/observability/event/event_observation_test.go @@ -4,7 +4,6 @@ import ( "testing" "time" - "github.com/hashicorp/eventlogger/filters/gated" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -40,7 +39,6 @@ func Test_newObservation(t *testing.T) { name: "valid-no-opts", fromOp: Op("valid-no-opts"), want: &observation{ - Payload: &gated.Payload{}, Version: errorVersion, Op: Op("valid-no-opts"), }, @@ -56,12 +54,10 @@ func Test_newObservation(t *testing.T) { WithFlush(), }, want: &observation{ - Payload: &gated.Payload{ - ID: "valid-all-opts", - Header: testHeader, - Detail: testDetails, - Flush: true, - }, + ID: "valid-all-opts", + Header: testHeader, + Detail: testDetails, + Flush: true, Version: errorVersion, Op: Op("valid-all-opts"), RequestInfo: TestRequestInfo(t), @@ -124,9 +120,7 @@ func Test_observationvalidate(t *testing.T) { assert, require := assert.New(t), require.New(t) e := observation{ Op: tt.op, - Payload: &gated.Payload{ - ID: tt.id, - }, + ID: tt.id, } err := e.validate() if tt.wantErrIs != nil { diff --git a/internal/observability/event/event_sys.go b/internal/observability/event/event_sys.go index 8fe4b76f8..3304fa358 100644 --- a/internal/observability/event/event_sys.go +++ b/internal/observability/event/event_sys.go @@ -11,4 +11,4 @@ type sysEvent struct { } // EventType is required for all event types by the eventlogger broker -func (e *sysEvent) EventType() string { return string(ErrorType) } +func (e *sysEvent) EventType() string { return string(SystemType) } diff --git a/internal/observability/event/event_sys_test.go b/internal/observability/event/event_sys_test.go new file mode 100644 index 000000000..ba5c84492 --- /dev/null +++ b/internal/observability/event/event_sys_test.go @@ -0,0 +1,14 @@ +package event + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func Test_SysEvent_EventType(t *testing.T) { + t.Run("simple", func(t *testing.T) { + e := sysEvent{} + require.Equal(t, string(SystemType), e.EventType()) + }) +} diff --git a/internal/observability/event/eventer.go b/internal/observability/event/eventer.go index 22e568086..94ad245a4 100644 --- a/internal/observability/event/eventer.go +++ b/internal/observability/event/eventer.go @@ -3,13 +3,16 @@ package event import ( "context" "fmt" + "net/url" "os" "sync" "time" "github.com/hashicorp/eventlogger" "github.com/hashicorp/eventlogger/filters/gated" + "github.com/hashicorp/eventlogger/formatter_filters/cloudevents" "github.com/hashicorp/eventlogger/sinks/writer" + "github.com/hashicorp/go-hclog" ) @@ -79,7 +82,7 @@ var ( // process should only have one Eventer. In practice this means the process // Server (Controller or Worker) and the SysEventer both need a pointer to a // single Eventer. -func InitSysEventer(log hclog.Logger, serializationLock *sync.Mutex, opt ...Option) error { +func InitSysEventer(log hclog.Logger, serializationLock *sync.Mutex, serverName string, opt ...Option) error { const op = "event.InitSysEventer" if log == nil { return fmt.Errorf("%s: missing hclog: %w", op, ErrInvalidParameter) @@ -87,6 +90,9 @@ func InitSysEventer(log hclog.Logger, serializationLock *sync.Mutex, opt ...Opti if serializationLock == nil { return fmt.Errorf("%s: missing serialization lock: %w", op, ErrInvalidParameter) } + if serverName == "" { + return fmt.Errorf("%s: missing server name: %w", op, ErrInvalidParameter) + } // the order of operations is important here. we want to determine if // there's an error before setting the singleton sysEventer @@ -101,7 +107,7 @@ func InitSysEventer(log hclog.Logger, serializationLock *sync.Mutex, opt ...Opti case opts.withEventerConfig != nil: var err error - if e, err = NewEventer(log, serializationLock, *opts.withEventerConfig); err != nil { + if e, err = NewEventer(log, serializationLock, serverName, *opts.withEventerConfig); err != nil { return fmt.Errorf("%s: %w", op, err) } @@ -125,7 +131,7 @@ func SysEventer() *Eventer { // NewEventer creates a new Eventer using the config. Supports options: // WithNow, WithSerializationLock, WithBroker -func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig, opt ...Option) (*Eventer, error) { +func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, serverName string, c EventerConfig, opt ...Option) (*Eventer, error) { const op = "event.NewEventer" if log == nil { return nil, fmt.Errorf("%s: missing logger: %w", op, ErrInvalidParameter) @@ -133,6 +139,9 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig if serializationLock == nil { return nil, fmt.Errorf("%s: missing serialization lock: %w", op, ErrInvalidParameter) } + if serverName == "" { + return nil, fmt.Errorf("%s: missing server name: %w", op, ErrInvalidParameter) + } // if there are no sinks in config, then we'll default to just one stderr // sink. @@ -165,18 +174,6 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig e.broker.StopTimeAt(opts.withNow) } - // Create JSONFormatter node - id, err := newId("json") - if err != nil { - return nil, fmt.Errorf("%s: %w", op, err) - } - jsonfmtId := eventlogger.NodeID(id) - fmtNode := &eventlogger.JSONFormatter{} - err = e.broker.RegisterNode(jsonfmtId, fmtNode) - if err != nil { - return nil, fmt.Errorf("%s: failed to register json node: %w", op, err) - } - // serializedStderr will be shared among all StderrSinks so their output is not // interwoven serializedStderr := serializedWriter{ @@ -189,6 +186,15 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig allSinkFilenames := map[string]bool{} for _, s := range c.Sinks { + fmtId, fmtNode, err := newFmtFilterNode(serverName, s) + if err != nil { + return nil, fmt.Errorf("%s: %w", op, err) + } + err = e.broker.RegisterNode(eventlogger.NodeID(fmtId), fmtNode) + if err != nil { + return nil, fmt.Errorf("%s: unable to register fmt/filter node: %w", op, err) + } + var sinkId eventlogger.NodeID var sinkNode eventlogger.Node switch s.SinkType { @@ -197,7 +203,7 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig Format: string(s.Format), Writer: &serializedStderr, } - id, err = newId("stderr") + id, err := NewId("stderr") if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } @@ -214,7 +220,7 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig MaxDuration: s.RotateDuration, MaxFiles: s.RotateMaxFiles, } - id, err = newId(fmt.Sprintf("file_%s_%s_", s.Path, s.FileName)) + id, err := NewId(fmt.Sprintf("file_%s_%s_", s.Path, s.FileName)) if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } @@ -245,7 +251,7 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig if addToAudit { auditPipelines = append(auditPipelines, pipeline{ eventType: AuditType, - fmtId: jsonfmtId, + fmtId: fmtId, sinkId: sinkId, sinkConfig: s, }) @@ -253,7 +259,7 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig if addToObservation { observationPipelines = append(observationPipelines, pipeline{ eventType: ObservationType, - fmtId: jsonfmtId, + fmtId: fmtId, sinkId: sinkId, sinkConfig: s, }) @@ -261,7 +267,7 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig if addToErr { errPipelines = append(errPipelines, pipeline{ eventType: ErrorType, - fmtId: jsonfmtId, + fmtId: fmtId, sinkId: sinkId, sinkConfig: s, }) @@ -269,7 +275,7 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig if addToSys { sysPipelines = append(sysPipelines, pipeline{ eventType: SystemType, - fmtId: jsonfmtId, + fmtId: fmtId, sinkId: sinkId, }) } @@ -284,12 +290,13 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig return nil, fmt.Errorf("%s: system events enabled but no sink defined for it: %w", op, ErrInvalidParameter) } + auditNodeIds := make([]eventlogger.NodeID, 0, len(auditPipelines)) for _, p := range auditPipelines { gatedFilterNode := gated.Filter{ Broker: e.broker, } e.flushableNodes = append(e.flushableNodes, &gatedFilterNode) - gateId, err := newId("gated-audit") + gateId, err := NewId("gated-audit") if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } @@ -298,7 +305,7 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig return nil, fmt.Errorf("%s: unable to register audit gated filter: %w", op, err) } - pipeId, err := newId(auditPipeline) + pipeId, err := NewId(auditPipeline) if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } @@ -310,14 +317,16 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig if err != nil { return nil, fmt.Errorf("%s: failed to register audit pipeline: %w", op, err) } + auditNodeIds = append(auditNodeIds, p.sinkId) } + observationNodeIds := make([]eventlogger.NodeID, 0, len(observationPipelines)) for _, p := range observationPipelines { gatedFilterNode := gated.Filter{ Broker: e.broker, } e.flushableNodes = append(e.flushableNodes, &gatedFilterNode) - gateId, err := newId("gated-observation") + gateId, err := NewId("gated-observation") if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } @@ -326,7 +335,7 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig return nil, fmt.Errorf("%s: unable to register audit gated filter: %w", op, err) } - pipeId, err := newId(observationPipeline) + pipeId, err := NewId(observationPipeline) if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } @@ -338,10 +347,11 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig if err != nil { return nil, fmt.Errorf("%s: failed to register observation pipeline: %w", op, err) } + observationNodeIds = append(observationNodeIds, p.sinkId) } errNodeIds := make([]eventlogger.NodeID, 0, len(errPipelines)) for _, p := range errPipelines { - pipeId, err := newId(errPipeline) + pipeId, err := NewId(errPipeline) if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } @@ -357,7 +367,7 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig } sysNodeIds := make([]eventlogger.NodeID, 0, len(sysPipelines)) for _, p := range sysPipelines { - pipeId, err := newId(sysPipeline) + pipeId, err := NewId(sysPipeline) if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } @@ -372,11 +382,22 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig sysNodeIds = append(sysNodeIds, p.sinkId) } - // always enforce delivery of errors + err := e.broker.SetSuccessThreshold(eventlogger.EventType(ObservationType), len(observationNodeIds)) + if err != nil { + return nil, fmt.Errorf("%s: failed to set success threshold for observation events: %w", op, err) + } + err = e.broker.SetSuccessThreshold(eventlogger.EventType(AuditType), len(auditNodeIds)) + if err != nil { + return nil, fmt.Errorf("%s: failed to set success threshold for audit events: %w", op, err) + } err = e.broker.SetSuccessThreshold(eventlogger.EventType(ErrorType), len(errNodeIds)) if err != nil { return nil, fmt.Errorf("%s: failed to set success threshold for error events: %w", op, err) } + err = e.broker.SetSuccessThreshold(eventlogger.EventType(SystemType), len(sysNodeIds)) + if err != nil { + return nil, fmt.Errorf("%s: failed to set success threshold for sysevents: %w", op, err) + } e.auditPipelines = append(e.auditPipelines, auditPipelines...) e.errPipelines = append(e.errPipelines, errPipelines...) @@ -385,6 +406,36 @@ func NewEventer(log hclog.Logger, serializationLock *sync.Mutex, c EventerConfig return e, nil } +func newFmtFilterNode(serverName string, c SinkConfig) (eventlogger.NodeID, eventlogger.Node, error) { + const op = "newFmtFilterNode" + if serverName == "" { + return "", nil, fmt.Errorf("%s: missing server name: %w", op, ErrInvalidParameter) + } + id, err := NewId("cloudevents") + if err != nil { + return "", nil, fmt.Errorf("%s: unable to generate id: %w", op, err) + } + var sourceUrl *url.URL + switch { + case c.EventSourceUrl != "": + sourceUrl, err = url.Parse(c.EventSourceUrl) + if err != nil { + return "", nil, fmt.Errorf("%s: invalid event source URL (%s): %w", op, c.EventSourceUrl, err) + } + default: + s := fmt.Sprintf("https://hashicorp.com/boundary/%s", serverName) + sourceUrl, err = url.Parse(s) + if err != nil { + return "", nil, fmt.Errorf("%s: invalid event source URL (%s): %w", op, s, err) + } + } + fmtNode, err := NewCloudEventsNode(sourceUrl, cloudevents.Format(c.Format), WithAllow(c.AllowFilters...), WithDeny(c.DenyFilters...)) + if err != nil { + return "", nil, fmt.Errorf("%s: %w", op, err) + } + return eventlogger.NodeID(id), fmtNode, nil +} + func DefaultEventerConfig() *EventerConfig { return &EventerConfig{ AuditEnabled: false, @@ -420,7 +471,7 @@ func (e *Eventer) writeObservation(ctx context.Context, event *observation) erro if event.Detail != nil { event.Detail[OpField] = string(event.Op) } - return e.broker.Send(ctx, eventlogger.EventType(ObservationType), event.Payload) + return e.broker.Send(ctx, eventlogger.EventType(ObservationType), event) }) if err != nil { e.logger.Error("encountered an error sending an observation event", "error:", err.Error()) diff --git a/internal/observability/event/eventer_retry_test.go b/internal/observability/event/eventer_retry_test.go index 01d5ee9e8..e593bf23d 100644 --- a/internal/observability/event/eventer_retry_test.go +++ b/internal/observability/event/eventer_retry_test.go @@ -24,7 +24,7 @@ func TestEventer_retrySend(t *testing.T) { testLogger := hclog.New(&hclog.LoggerOptions{ Mutex: testLock, }) - eventer, err := NewEventer(testLogger, testLock, testConfig.EventerConfig) + eventer, err := NewEventer(testLogger, testLock, "TestEventer_retrySend", testConfig.EventerConfig) require.NoError(t, err) testError := fmt.Errorf("%s: missing operation: %w", "missing operation", ErrInvalidParameter) diff --git a/internal/observability/event/eventer_test.go b/internal/observability/event/eventer_test.go index 0ba6e8e32..27d6ae4b5 100644 --- a/internal/observability/event/eventer_test.go +++ b/internal/observability/event/eventer_test.go @@ -21,48 +21,84 @@ func Test_InitSysEventer(t *testing.T) { testLogger := hclog.New(&hclog.LoggerOptions{ Mutex: testLock, }) - testEventer, err := NewEventer(testLogger, testLock, testConfig.EventerConfig) + testEventer, err := NewEventer(testLogger, testLock, "Test_InitSysEventer", testConfig.EventerConfig) require.NoError(t, err) tests := []struct { - name string - log hclog.Logger - lock *sync.Mutex - opt []Option - want *Eventer - wantErrIs error + name string + log hclog.Logger + lock *sync.Mutex + serverName string + opt []Option + want *Eventer + wantErrIs error }{ { - name: "missing-both-eventer-and-config", - wantErrIs: ErrInvalidParameter, + name: "missing-both-eventer-and-config", + serverName: "missing-both-eventer-and-config", + wantErrIs: ErrInvalidParameter, }, { - name: "missing-hclog", + name: "missing-hclog", + opt: []Option{WithEventerConfig(&testConfig.EventerConfig)}, + lock: testLock, + serverName: "missing-hclog", + wantErrIs: ErrInvalidParameter, + }, + { + name: "missing-lock", + opt: []Option{WithEventerConfig(&testConfig.EventerConfig)}, + log: testLogger, + serverName: "missing-lock", + wantErrIs: ErrInvalidParameter, + }, + { + name: "missing-serverName", opt: []Option{WithEventerConfig(&testConfig.EventerConfig)}, + log: testLogger, lock: testLock, wantErrIs: ErrInvalidParameter, }, { - name: "missing-lock", - opt: []Option{WithEventerConfig(&testConfig.EventerConfig)}, - log: testLogger, - wantErrIs: ErrInvalidParameter, + name: "missing-eventer-and-config", + log: testLogger, + lock: testLock, + serverName: "success-with-config", + wantErrIs: ErrInvalidParameter, }, { - name: "success-with-config", - opt: []Option{WithEventerConfig(&testConfig.EventerConfig)}, - log: testLogger, - lock: testLock, + name: "both-eventer-and-config", + opt: []Option{WithEventerConfig(&testConfig.EventerConfig), WithEventer(testEventer)}, + log: testLogger, + lock: testLock, + serverName: "success-with-config", + wantErrIs: ErrInvalidParameter, + }, + { + name: "bad-config", + opt: []Option{WithEventerConfig(&EventerConfig{Sinks: []SinkConfig{{Format: "bad-format"}}})}, + log: testLogger, + lock: testLock, + serverName: "success-with-config", + wantErrIs: ErrInvalidParameter, + }, + { + name: "success-with-config", + opt: []Option{WithEventerConfig(&testConfig.EventerConfig)}, + log: testLogger, + lock: testLock, + serverName: "success-with-config", want: &Eventer{ logger: testLogger, conf: testConfig.EventerConfig, }, }, { - name: "success-with-default-config", - opt: []Option{WithEventerConfig(&EventerConfig{})}, - log: testLogger, - lock: testLock, + name: "success-with-default-config", + opt: []Option{WithEventerConfig(&EventerConfig{})}, + log: testLogger, + lock: testLock, + serverName: "success-with-default-config", want: &Eventer{ logger: testLogger, conf: EventerConfig{ @@ -78,11 +114,12 @@ func Test_InitSysEventer(t *testing.T) { }, }, { - name: "success-with-eventer", - opt: []Option{WithEventer(testEventer)}, - log: testLogger, - lock: testLock, - want: testEventer, + name: "success-with-eventer", + opt: []Option{WithEventer(testEventer)}, + log: testLogger, + lock: testLock, + serverName: "success-with-eventer", + want: testEventer, }, } for _, tt := range tests { @@ -90,7 +127,7 @@ func Test_InitSysEventer(t *testing.T) { defer TestResetSystEventer(t) assert, require := assert.New(t), require.New(t) - err := InitSysEventer(tt.log, tt.lock, tt.opt...) + err := InitSysEventer(tt.log, tt.lock, tt.serverName, tt.opt...) got := SysEventer() if tt.wantErrIs != nil { require.Nil(got) @@ -118,7 +155,7 @@ func TestEventer_writeObservation(t *testing.T) { testLogger := hclog.New(&hclog.LoggerOptions{ Mutex: testLock, }) - eventer, err := NewEventer(testLogger, testLock, testSetup.EventerConfig) + eventer, err := NewEventer(testLogger, testLock, "TestEventer_writeObservation", testSetup.EventerConfig) require.NoError(t, err) testHeader := map[string]interface{}{"name": "header"} @@ -176,7 +213,7 @@ func TestEventer_writeObservation(t *testing.T) { Name: "test", }) // with no defined config, it will default to a stderr sink - e, err := NewEventer(testLogger, testLock, c) + e, err := NewEventer(testLogger, testLock, "e2e-test", c) require.NoError(err) m := map[string]interface{}{ @@ -199,7 +236,7 @@ func TestEventer_writeAudit(t *testing.T) { Mutex: testLock, Name: "test", }) - eventer, err := NewEventer(testLogger, testLock, testSetup.EventerConfig) + eventer, err := NewEventer(testLogger, testLock, "TestEventer_writeAudit", testSetup.EventerConfig) require.NoError(t, err) testAudit, err := newAudit( @@ -259,7 +296,7 @@ func TestEventer_writeError(t *testing.T) { Mutex: testLock, Name: "test", }) - eventer, er := NewEventer(testLogger, testLock, testSetup.EventerConfig) + eventer, er := NewEventer(testLogger, testLock, "TestEventer_writeError", testSetup.EventerConfig) require.NoError(t, er) testError, er := newError("TestEventer_writeError", fmt.Errorf("%s: no msg: test", ErrIo)) @@ -323,6 +360,7 @@ func Test_NewEventer(t *testing.T) { opts []Option logger hclog.Logger lock *sync.Mutex + serverName string want *Eventer wantRegistered []string wantPipelines []string @@ -330,22 +368,32 @@ func Test_NewEventer(t *testing.T) { wantErrIs error }{ { - name: "missing-logger", + name: "missing-logger", + config: testSetup.EventerConfig, + lock: testLock, + serverName: "missing-logger", + wantErrIs: ErrInvalidParameter, + }, + { + name: "missing-lock", + config: testSetup.EventerConfig, + logger: testLogger, + serverName: "missing-lock", + wantErrIs: ErrInvalidParameter, + }, + { + name: "missing-server-name", config: testSetup.EventerConfig, + logger: testLogger, lock: testLock, wantErrIs: ErrInvalidParameter, }, { - name: "missing-lock", - config: testSetup.EventerConfig, - logger: testLogger, - wantErrIs: ErrInvalidParameter, - }, - { - name: "success-with-default-config", - config: EventerConfig{}, - logger: testLogger, - lock: testLock, + name: "success-with-default-config", + config: EventerConfig{}, + logger: testLogger, + lock: testLock, + serverName: "success-with-default-config", want: &Eventer{ logger: testLogger, conf: EventerConfig{ @@ -360,7 +408,7 @@ func Test_NewEventer(t *testing.T) { }, }, wantRegistered: []string{ - "json", // fmt for everything + "cloudevents", // fmt for everything "stderr", // stderr "gated-observation", // stderr "gated-audit", // stderr @@ -372,26 +420,32 @@ func Test_NewEventer(t *testing.T) { "system", // stderr }, wantThresholds: map[eventlogger.EventType]int{ - "error": 1, + "error": 1, + "system": 1, + "observation": 1, + "audit": 1, }, }, { - name: "testSetup", - config: testSetup.EventerConfig, - logger: testLogger, - lock: testLock, + name: "testSetup", + config: testSetup.EventerConfig, + logger: testLogger, + lock: testLock, + serverName: "testSetup", want: &Eventer{ logger: testLogger, conf: testSetup.EventerConfig, }, wantRegistered: []string{ - "json", // fmt for everything + "cloudevents", // stderr "stderr", // stderr "gated-observation", // stderr "gated-audit", // stderr + "cloudevents", // every-type-file-sync "tmp-all-events", // every-type-file-sync "gated-observation", // every-type-file-sync "gated-audit", // every-type-file-sync + "cloudevents", // error-file-sink "tmp-errors", // error-file-sink }, wantPipelines: []string{ @@ -406,31 +460,40 @@ func Test_NewEventer(t *testing.T) { "system", // stderr }, wantThresholds: map[eventlogger.EventType]int{ - "error": 3, + "error": 3, + "system": 2, + "observation": 2, + "audit": 2, }, }, { - name: "testSetup-with-all-opts", - config: testSetupWithOpts.EventerConfig, - logger: testLogger, - lock: testLock, + name: "testSetup-with-all-opts", + config: testSetupWithOpts.EventerConfig, + logger: testLogger, + lock: testLock, + serverName: "testSetup-with-all-opts", want: &Eventer{ logger: testLogger, conf: testSetupWithOpts.EventerConfig, }, wantRegistered: []string{ - "json", // fmt for everything + "cloudevents", // stderr "stderr", // stderr "gated-observation", // stderr "gated-audit", // stderr + "cloudevents", // every-type-file-sync "tmp-all-events", // every-type-file-sync "gated-observation", // every-type-file-sync "gated-audit", // every-type-file-sync + "cloudevents", // error-file-sink "tmp-errors", // error-file-sink + "cloudevents", // observation-file-sink "gated-observation", // observation-file-sink "tmp-observation", // observations-file-sink + "cloudevents", // audit-file-sink "gated-audit", // audit-file-sink "tmp-audit", // audit-file-sink + "cloudevents", // sys-file-sink "tmp-sysevents", // sys-file-sink }, wantPipelines: []string{ @@ -448,7 +511,10 @@ func Test_NewEventer(t *testing.T) { "system", // sys-file-sink }, wantThresholds: map[eventlogger.EventType]int{ - "error": 3, + "error": 3, + "system": 3, + "observation": 3, + "audit": 3, }, }, } @@ -457,7 +523,7 @@ func Test_NewEventer(t *testing.T) { t.Run(tt.name, func(t *testing.T) { assert, require := assert.New(t), require.New(t) testBroker := &testMockBroker{} - got, err := NewEventer(tt.logger, tt.lock, tt.config, TestWithBroker(t, testBroker)) + got, err := NewEventer(tt.logger, tt.lock, tt.serverName, tt.config, TestWithBroker(t, testBroker)) if tt.wantErrIs != nil { require.Error(err) require.Nil(got) @@ -548,7 +614,7 @@ func TestEventer_Reopen(t *testing.T) { Name: "test", }) - e, err := NewEventer(testLogger, testLock, EventerConfig{}) + e, err := NewEventer(testLogger, testLock, "TestEventer_Reopen", EventerConfig{}) require.NoError(err) e.broker = nil @@ -570,7 +636,7 @@ func TestEventer_FlushNodes(t *testing.T) { Name: "test", }) - e, err := NewEventer(testLogger, testLock, EventerConfig{}) + e, err := NewEventer(testLogger, testLock, "TestEventer_FlushNodes", EventerConfig{}) require.NoError(err) node := &testFlushNode{} diff --git a/internal/observability/event/id.go b/internal/observability/event/id.go index fc4b70667..239f13d31 100644 --- a/internal/observability/event/id.go +++ b/internal/observability/event/id.go @@ -6,10 +6,12 @@ import ( "github.com/hashicorp/go-secure-stdlib/base62" ) -// This modified newId has been done to stop a circular +const idPrefix = "e" + +// NewId is a bit of a modified NewId has been done to stop a circular // dependency with the errors package that is caused by importing // boundary/internal/db -func newId(prefix string) (string, error) { +func NewId(prefix string) (string, error) { const op = "event.newId" if prefix == "" { return "", fmt.Errorf("%s: missing prefix: %w", op, ErrInvalidParameter) diff --git a/internal/observability/event/id_test.go b/internal/observability/event/id_test.go index f634371cd..0c39ea4e7 100644 --- a/internal/observability/event/id_test.go +++ b/internal/observability/event/id_test.go @@ -11,7 +11,7 @@ import ( func Test_newId(t *testing.T) { t.Run("basics", func(t *testing.T) { assert, require := assert.New(t), require.New(t) - got, err := newId("pre") + got, err := NewId("pre") require.NoError(err) assert.True(strings.HasPrefix(got, "pre")) }) diff --git a/internal/observability/event/node.go b/internal/observability/event/node.go new file mode 100644 index 000000000..8fd226f07 --- /dev/null +++ b/internal/observability/event/node.go @@ -0,0 +1,126 @@ +package event + +import ( + "context" + "fmt" + "net/url" + + filterpkg "github.com/hashicorp/boundary/internal/filter" + "github.com/hashicorp/eventlogger" + "github.com/hashicorp/eventlogger/formatter_filters/cloudevents" + "github.com/hashicorp/go-bexpr" +) + +// Node represents an eventlogger.Node which filters events based on allow and +// deny bexpr filters +type Node struct { + *cloudevents.FormatterFilter + allow []*filter + deny []*filter +} + +// NewCloudEventsNode creates a new filter node using the optional allow and deny filters +// provided. Support for WithAllow and WithDeny options. +func NewCloudEventsNode(source *url.URL, format cloudevents.Format, opt ...Option) (*Node, error) { + const op = "event.NewCloudEventsNode" + if source == nil { + return nil, fmt.Errorf("%s: missing source: %w", op, ErrInvalidParameter) + } + switch format { + case cloudevents.FormatJSON, cloudevents.FormatText: + default: + return nil, fmt.Errorf("%s: invalid format '%s': %w", op, format, ErrInvalidParameter) + } + opts := getOpts(opt...) + n := Node{ + FormatterFilter: &cloudevents.FormatterFilter{ + Source: source, + Schema: opts.withSchema, + Format: format, + }, + } + + // intentionally not checking if allow and/or deny optional filters were + // supplied since having a filter node with no filters is okay. + + if len(opts.withAllow) > 0 { + n.allow = make([]*filter, 0, len((opts.withAllow))) + for i := range opts.withAllow { + f, err := newFilter(opts.withAllow[i]) + if err != nil { + return nil, fmt.Errorf("%s: invalid allow filter '%s': %w", op, opts.withAllow[i], err) + } + n.allow = append(n.allow, f) + } + } + if len(opts.withDeny) > 0 { + n.deny = make([]*filter, 0, len((opts.withDeny))) + for i := range opts.withDeny { + f, err := newFilter(opts.withDeny[i]) + if err != nil { + return nil, fmt.Errorf("%s: invalid deny filter '%s': %w", op, opts.withDeny[i], err) + } + n.deny = append(n.deny, f) + } + } + n.Predicate = newPredicate(n.allow, n.deny) + return &n, nil +} + +func newPredicate(allow, deny []*filter) func(ctx context.Context, ce interface{}) (bool, error) { + return func(ctx context.Context, ce interface{}) (bool, error) { + if len(allow) == 0 && len(deny) == 0 { + return true, nil + } + for _, f := range deny { + if f.Match(ce) { + return false, nil + } + } + switch { + case len(allow) > 0: + for _, f := range allow { + if f.Match(ce) { + return true, nil + } + } + return false, nil + default: + return true, nil + } + } +} + +var _ eventlogger.Node = &Node{} + +type filter struct { + raw string + eval *bexpr.Evaluator +} + +// newFilter returns a Filter which can be matched against. +func newFilter(f string) (*filter, error) { + const op = "event.newFilter" + if f == "" { + return nil, fmt.Errorf("%s: missing filter: %w", op, ErrInvalidParameter) + } + e, err := bexpr.CreateEvaluator(f, bexpr.WithHookFn(filterpkg.WellKnownTypeFilterHook)) + if err != nil { + return nil, fmt.Errorf("%s: %w", op, err) + } + return &filter{eval: e, raw: f}, nil +} + +// Match returns if the provided interface matches the filter. If the filter +// does not match the structure of the object being Matched, false is returned. +func (f *filter) Match(item interface{}) bool { + if f.eval == nil { + return true + } + m, err := f.eval.Evaluate(item) + // There isn't a clear way to differentiate between a JSON Pointer which doesn't represent + // the structure of the object being Matched and a JSON Pointer which references a field which + // is part of a sub structure that is nil in this item. Because of this, any filter which would + // result in an error using the underlying library is simply interpreted as not a match. + return err == nil && m +} diff --git a/internal/observability/event/node_test.go b/internal/observability/event/node_test.go new file mode 100644 index 000000000..6f74e909c --- /dev/null +++ b/internal/observability/event/node_test.go @@ -0,0 +1,346 @@ +package event + +import ( + "context" + "encoding/json" + "fmt" + "net/url" + "testing" + "time" + + "github.com/hashicorp/eventlogger" + "github.com/hashicorp/eventlogger/formatter_filters/cloudevents" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func Test_NewCloudEventsNode(t *testing.T) { + t.Parallel() + testSource, err := url.Parse("https://localhost:9200") + require.NoError(t, err) + tests := []struct { + name string + source *url.URL + format cloudevents.Format + opt []Option + wantErr bool + wantIsError error + wantErrContains string + wantAllow []string + wantDeny []string + }{ + { + name: "no-opts", + source: testSource, + format: cloudevents.FormatJSON, + }, + { + name: "bad-allow-filter", + source: testSource, + format: cloudevents.FormatJSON, + opt: []Option{ + WithAllow("foo=;22", "foo==bar"), + }, + wantErr: true, + wantErrContains: "invalid allow filter 'foo=;22'", + }, + { + name: "bad-deny-filter", + source: testSource, + format: cloudevents.FormatJSON, + opt: []Option{ + WithDeny("foo=;22", "foo==bar"), + }, + wantErr: true, + wantErrContains: "invalid deny filter 'foo=;22'", + }, + { + name: "empty-allow-filter", + source: testSource, + format: cloudevents.FormatJSON, + opt: []Option{ + WithAllow(""), + }, + wantErr: true, + wantErrContains: "missing filter", + }, + { + name: "empty-deny-filter", + source: testSource, + format: cloudevents.FormatJSON, + opt: []Option{ + WithDeny(""), + }, + wantErr: true, + wantErrContains: "missing filter", + }, + { + name: "empty-source", + format: cloudevents.FormatJSON, + opt: []Option{ + WithAllow("alice==friend", "bob==friend"), + WithDeny("eve==acquaintance", "fido!=dog"), + }, + wantErr: true, + wantErrContains: "missing source", + }, + { + name: "bad-format", + source: testSource, + format: "invalid-format", + opt: []Option{ + WithAllow("alice==friend", "bob==friend"), + WithDeny("eve==acquaintance", "fido!=dog"), + }, + wantErr: true, + wantErrContains: "invalid format", + }, + { + name: "valid-filters", + source: testSource, + format: cloudevents.FormatJSON, + opt: []Option{ + WithAllow("alice==friend", "bob==friend"), + WithDeny("eve==acquaintance", "fido!=dog"), + }, + wantAllow: []string{"alice==friend", "bob==friend"}, + wantDeny: []string{"eve==acquaintance", "fido!=dog"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + got, err := NewCloudEventsNode(tt.source, tt.format, tt.opt...) + if tt.wantErr { + require.Error(err) + assert.Nil(got) + if tt.wantIsError != nil { + assert.ErrorIs(err, tt.wantIsError) + } + if tt.wantErrContains != "" { + assert.Contains(err.Error(), tt.wantErrContains) + } + return + } + require.NoError(err) + assert.NotNil(got) + assert.Len(got.allow, len(tt.wantAllow)) + for _, f := range got.allow { + assert.Contains(tt.wantAllow, f.raw) + } + assert.Len(got.deny, len(tt.wantDeny)) + for _, f := range got.deny { + assert.Contains(tt.wantDeny, f.raw) + } + + }) + } +} + +func TestNode_Process(t *testing.T) { + t.Parallel() + ctx := context.Background() + testUrl, err := url.Parse("https://localhost") + require.NoError(t, err) + now := time.Now() + + testNode, err := NewCloudEventsNode(testUrl, cloudevents.FormatJSON, WithSchema(testUrl)) + require.NoError(t, err) + + f, err := newFilter(`Data == "match-filter"`) + require.NoError(t, err) + + tests := []struct { + name string + n *Node + e *eventlogger.Event + format cloudevents.Format + predicate func(ctx context.Context, ce interface{}) (bool, error) + wantCloudEvent *cloudevents.Event + wantText string + wantIsError error + wantErrContains string + }{ + { + name: "simple-JSON", + n: testNode, + e: &eventlogger.Event{ + Type: "test", + CreatedAt: now, + Payload: "test-string", + }, + format: cloudevents.FormatJSON, + wantCloudEvent: &cloudevents.Event{ + Source: testUrl.String(), + DataSchema: testUrl.String(), + SpecVersion: cloudevents.SpecVersion, + Type: "test", + Data: "test-string", + DataContentType: "application/cloudevents", + Time: now, + }, + }, + { + name: "deny-filter-match", + n: testNode, + e: &eventlogger.Event{ + Type: "test", + CreatedAt: now, + Payload: "match-filter", + }, + format: cloudevents.FormatJSON, + predicate: newPredicate(nil, []*filter{f}), + }, + { + name: "deny-filter-not-matching", + n: testNode, + e: &eventlogger.Event{ + Type: "test", + CreatedAt: now, + Payload: "not-matching-filter", + }, + format: cloudevents.FormatJSON, + predicate: newPredicate(nil, []*filter{f}), + wantCloudEvent: &cloudevents.Event{ + Source: testUrl.String(), + DataSchema: testUrl.String(), + SpecVersion: cloudevents.SpecVersion, + Type: "test", + Data: "not-matching-filter", + DataContentType: "application/cloudevents", + Time: now, + }, + }, + { + name: "allow-filter-match", + n: testNode, + e: &eventlogger.Event{ + Type: "test", + CreatedAt: now, + Payload: "match-filter", + }, + format: cloudevents.FormatJSON, + predicate: newPredicate([]*filter{f}, nil), + wantCloudEvent: &cloudevents.Event{ + Source: testUrl.String(), + DataSchema: testUrl.String(), + SpecVersion: cloudevents.SpecVersion, + Type: "test", + Data: "match-filter", + DataContentType: "application/cloudevents", + Time: now, + }, + }, + { + name: "allow-filter-not-matching", + n: testNode, + e: &eventlogger.Event{ + Type: "test", + CreatedAt: now, + Payload: "not-matching-filter", + }, + format: cloudevents.FormatJSON, + predicate: newPredicate([]*filter{f}, nil), + }, + { + name: "no-filters", + n: testNode, + e: &eventlogger.Event{ + Type: "test", + CreatedAt: now, + Payload: "test-data", + }, + format: cloudevents.FormatJSON, + predicate: newPredicate(nil, nil), + wantCloudEvent: &cloudevents.Event{ + Source: testUrl.String(), + DataSchema: testUrl.String(), + SpecVersion: cloudevents.SpecVersion, + Type: "test", + Data: "test-data", + DataContentType: "application/cloudevents", + Time: now, + }, + }, + { + name: "simple-Text", + n: testNode, + e: &eventlogger.Event{ + Type: "test", + CreatedAt: now, + Payload: "test-string", + }, + format: cloudevents.FormatText, + wantCloudEvent: &cloudevents.Event{ + Source: testUrl.String(), + DataSchema: testUrl.String(), + SpecVersion: cloudevents.SpecVersion, + Type: "test", + Data: "test-string", + DataContentType: "text/plain", + Time: now, + }, + wantText: `{ + "id": "%s", + "source": "https://localhost", + "specversion": "1.0", + "type": "test", + "data": "test-string", + "datacontentype": "text/plain", + "dataschema": "https://localhost", + "time": %s +} +`, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // these subtests cannot be run in parallel since they all depend on + // a shared testNode which may be modified during the test. + assert, require := assert.New(t), require.New(t) + tt.n.Format = tt.format + tt.n.Predicate = tt.predicate + + gotEvent, err := tt.n.Process(ctx, tt.e) + if tt.wantIsError != nil { + require.Error(err) + assert.Nil(gotEvent) + assert.ErrorIs(err, tt.wantIsError) + if tt.wantErrContains != "" { + assert.Contains(err.Error(), tt.wantErrContains) + } + return + } + if tt.wantCloudEvent == nil { + assert.Nil(gotEvent) + return + } + gotFormatted, ok := gotEvent.Format(string(tt.format)) + require.True(ok) + var gotCloudEvent cloudevents.Event + require.NoError(json.Unmarshal(gotFormatted, &gotCloudEvent)) + if tt.wantCloudEvent.ID == "" { + tt.wantCloudEvent.ID = gotCloudEvent.ID + } + var wantJSON []byte + switch tt.format { + case cloudevents.FormatJSON: + wantJSON, err = json.Marshal(tt.wantCloudEvent) + case cloudevents.FormatText: + // test the raw JSON + jsonTime, err := gotCloudEvent.Time.MarshalJSON() + require.NoError(err) + wantRawText := []byte(fmt.Sprintf(tt.wantText, gotCloudEvent.ID, jsonTime)) + assert.Equal(string(wantRawText), string(gotFormatted)) + + // test the marshaled JSON + wantJSON, err = json.MarshalIndent(tt.wantCloudEvent, cloudevents.TextIndent, cloudevents.TextIndent) + require.NoError(err) + } + require.NoError(err) + assert.JSONEq(string(wantJSON), string(gotFormatted)) + t.Log(string(gotFormatted)) + }) + } +} diff --git a/internal/observability/event/options.go b/internal/observability/event/options.go index 977579c90..fddcbceab 100644 --- a/internal/observability/event/options.go +++ b/internal/observability/event/options.go @@ -1,6 +1,7 @@ package event import ( + "net/url" "time" ) @@ -31,6 +32,9 @@ type options struct { withAuth *Auth withEventer *Eventer withEventerConfig *EventerConfig + withAllow []string + withDeny []string + withSchema *url.URL withBroker broker // test only option withAuditSink bool // test only option @@ -118,3 +122,24 @@ func WithEventerConfig(c *EventerConfig) Option { o.withEventerConfig = c } } + +// WithSchema is an optional schema for the cloudevents +func WithSchema(url *url.URL) Option { + return func(o *options) { + o.withSchema = url + } +} + +// WithAllow is an optional set of allow filters +func WithAllow(f ...string) Option { + return func(o *options) { + o.withAllow = f + } +} + +// WithDeny is an optional set of deny filters +func WithDeny(f ...string) Option { + return func(o *options) { + o.withDeny = f + } +} diff --git a/internal/observability/event/options_test.go b/internal/observability/event/options_test.go index 6fefd0960..9549b6a4f 100644 --- a/internal/observability/event/options_test.go +++ b/internal/observability/event/options_test.go @@ -1,10 +1,12 @@ package event import ( + "net/url" "testing" "time" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) // Test_GetOpts provides unit tests for GetOpts and all the options @@ -100,4 +102,29 @@ func Test_GetOpts(t *testing.T) { testOpts.withEventerConfig = &c assert.Equal(opts, testOpts) }) + t.Run("WithAllow", func(t *testing.T) { + assert := assert.New(t) + allow := []string{"foo == bar", "bar == foo"} + opts := getOpts(WithAllow(allow...)) + testOpts := getDefaultOptions() + testOpts.withAllow = allow + assert.Equal(opts, testOpts) + }) + t.Run("WithDeny", func(t *testing.T) { + assert := assert.New(t) + deny := []string{"foo == bar", "bar == foo"} + opts := getOpts(WithDeny(deny...)) + testOpts := getDefaultOptions() + testOpts.withDeny = deny + assert.Equal(opts, testOpts) + }) + t.Run("WithSchema", func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + schema, err := url.Parse("https://alice.com") + require.NoError(err) + opts := getOpts(WithSchema(schema)) + testOpts := getDefaultOptions() + testOpts.withSchema = schema + assert.Equal(opts, testOpts) + }) } diff --git a/internal/observability/event/sink_config.go b/internal/observability/event/sink_config.go index 12e680e0d..a3b8bbbb2 100644 --- a/internal/observability/event/sink_config.go +++ b/internal/observability/event/sink_config.go @@ -10,8 +10,11 @@ type SinkConfig struct { Name string `hcl:"name"` // Name defines a name for the sink. Description string `hcl:"description"` // Description defines a description for the sink. EventTypes []Type `hcl:"event_types"` // EventTypes defines a list of event types that will be sent to the sink. See the docs for EventTypes for a list of accepted values. + EventSourceUrl string `hcl:"event_source_url"` // EventSource defines an optional event source URL for the sink. If not defined a default source will be composed of the https://hashicorp.com/boundary.io/ServerName/Path/FileName + AllowFilters []string `hcl:"allow_filters"` // AllowFilters define a set predicates for including an event in the sink. If any filter matches, the event will be included. The filter should be in a format supported by hashicorp/go-bexpr. + DenyFilters []string `hcl:"deny_filters"` // DenyFilters define a set predicates for excluding an event in the sink. If any filter matches, the event will be excluded. The filter should be in a format supported by hashicorp/go-bexpr. SinkType SinkType `hcl:"sink_type"` // SinkType defines the type of sink (StderrSink or FileSink) - Format SinkFormat `hcl:"format"` // Format defines the format for the sink (JSONSinkFormat) + Format SinkFormat `hcl:"format"` // Format defines the format for the sink (JSONSinkFormat or TextSinkFormat) Path string `hcl:"path"` // Path defines the file path for the sink FileName string `hcl:"file_name"` // FileName defines the file name for the sink RotateBytes int `hcl:"rotate_bytes"` // RotateByes defines the number of bytes that should trigger rotation of a FileSink @@ -43,3 +46,32 @@ func (sc *SinkConfig) validate() error { } return nil } + +// FilterType defines a type for filters (allow or deny) +type FilterType string + +const ( + AllowFilter FilterType = "allow" // AllowFilter defines a filter type for "allow" + DenyFilter FilterType = "deny" // DenyFilter defines a filter type for "deny" +) + +// SinkFilter defines an event filter (allow or deny) for a sink +type SinkFilter struct { + Type FilterType `hcl:"type"` // Type of filter (allow or deny) + Filter string `hcl:"filter"` // Filter in a format supported by hashicorp/go-bexpr. +} + +// Validate a SinkFilter +func (s SinkFilter) Validate() error { + const op = "event.(SinkFilter).Validate" + switch s.Type { + case AllowFilter, DenyFilter: + default: + return fmt.Errorf("%s: invalid filter type %s: %w", op, s.Type, ErrInvalidParameter) + } + _, err := newFilter(s.Filter) + if err != nil { + return fmt.Errorf("%s: invalid filter '%s': %w", op, s.Filter, err) + } + return nil +} diff --git a/internal/observability/event/sink_format.go b/internal/observability/event/sink_format.go index 5712f8c66..771f810e5 100644 --- a/internal/observability/event/sink_format.go +++ b/internal/observability/event/sink_format.go @@ -5,7 +5,8 @@ import ( ) const ( - JSONSinkFormat SinkFormat = "json" // JSONSinkFormat means the event is formatted as JSON + JSONSinkFormat SinkFormat = "cloudevents-json" // JSONSinkFormat means the event is formatted as JSON + TextSinkFormat SinkFormat = "cloudevents-text" // TextSinkFormat means the event is formmatted as text ) type SinkFormat string // SinkFormat defines the formatting for a sink in a config file stanza (json) @@ -13,7 +14,7 @@ type SinkFormat string // SinkFormat defines the formatting for a sink in a conf func (f SinkFormat) Validate() error { const op = "event.(SinkFormat).Validate" switch f { - case JSONSinkFormat: + case JSONSinkFormat, TextSinkFormat: return nil default: return fmt.Errorf("%s: '%s' is not a valid sink format: %w", op, f, ErrInvalidParameter) diff --git a/internal/observability/event/testing.go b/internal/observability/event/testing.go index ae1a7fd6c..4bfb5b701 100644 --- a/internal/observability/event/testing.go +++ b/internal/observability/event/testing.go @@ -154,6 +154,7 @@ func TestEventerConfig(t *testing.T, testName string, opt ...Option) TestConfig func TestRequestInfo(t *testing.T) *RequestInfo { t.Helper() return &RequestInfo{ + EventId: "test-event-id", Id: "test-request-info", Method: "POST", Path: "/test/request/info", diff --git a/internal/servers/common/handler.go b/internal/servers/common/handler.go index cdaba93e5..f5975178f 100644 --- a/internal/servers/common/handler.go +++ b/internal/servers/common/handler.go @@ -122,7 +122,14 @@ func WrapWithEventsHandler(h http.Handler, e *event.Eventer, logger hclog.Logger publicId, _, _ := auth.GetTokenFromRequest(logger, kms, r) var err error + id, err := event.NewId("e") + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + logger.Trace("unable to create id for event", "method", r.Method, "url", r.URL.RequestURI(), "error", err) + return + } info := &event.RequestInfo{ + EventId: id, Id: generatedTraceId(ctx), PublicId: publicId, Method: r.Method, diff --git a/internal/servers/common/handler_test.go b/internal/servers/common/handler_test.go index 0b410d3a4..10c7794c7 100644 --- a/internal/servers/common/handler_test.go +++ b/internal/servers/common/handler_test.go @@ -20,6 +20,7 @@ import ( "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/eventlogger" "github.com/hashicorp/eventlogger/filters/gated" + "github.com/hashicorp/eventlogger/formatter_filters/cloudevents" "github.com/hashicorp/go-hclog" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -180,7 +181,7 @@ func Test_WrapWithEventsHandler(t *testing.T) { Mutex: testLock, Name: "test", }) - testEventer, err := event.NewEventer(testLogger, testLock, c.EventerConfig) + testEventer, err := event.NewEventer(testLogger, testLock, "Test_WrapWithEventsHandler", c.EventerConfig) require.NoError(t, err) tests := []struct { @@ -234,7 +235,7 @@ func Test_WrapWithEventsHandler(t *testing.T) { e: func() *event.Eventer { b := &testMockBroker{errorOnSendAudit: true} c := event.EventerConfig{AuditEnabled: true} - e, err := event.NewEventer(testLogger, testLock, c, event.TestWithBroker(t, b)) + e, err := event.NewEventer(testLogger, testLock, "audit-startGatedEvents", c, event.TestWithBroker(t, b)) require.NoError(t, err) return e }(), @@ -249,7 +250,7 @@ func Test_WrapWithEventsHandler(t *testing.T) { e: func() *event.Eventer { b := &testMockBroker{errorOnFlush: true} c := event.EventerConfig{AuditEnabled: true} - e, err := event.NewEventer(testLogger, testLock, c, event.TestWithBroker(t, b)) + e, err := event.NewEventer(testLogger, testLock, "audit-flushGatedEvents", c, event.TestWithBroker(t, b)) require.NoError(t, err) return e }(), @@ -299,7 +300,7 @@ func Test_WrapWithEventsHandler(t *testing.T) { assert.Lenf(b, 0, "expected no json for internal errors but got %s", string(b)) return } - got := &eventJson{} + got := &cloudevents.Event{} err = json.Unmarshal(b, got) require.NoErrorf(err, "json: %s", string(b)) @@ -311,13 +312,13 @@ func Test_WrapWithEventsHandler(t *testing.T) { info := event.RequestInfo{ Method: "GET", Path: "/greeting", - Id: got.Payload["id"].(string), + Id: got.Data.(map[string]interface{})["request_info"].(map[string]interface{})["id"].(string), } hdr := map[string]interface{}{ "status": http.StatusTeapot, - "start": got.Payload["header"].(map[string]interface{})["start"].(string), - "stop": got.Payload["header"].(map[string]interface{})["stop"].(string), - "latency-ms": got.Payload["header"].(map[string]interface{})["latency-ms"].(float64), + "start": got.Data.(map[string]interface{})["start"].(string), + "stop": got.Data.(map[string]interface{})["stop"].(string), + "latency-ms": got.Data.(map[string]interface{})["latency-ms"].(float64), } wantJson := testJson(t, event.ObservationType, &info, event.Op(tt.name), got, hdr, nil) assert.JSONEq(string(wantJson), string(actualJson)) @@ -329,7 +330,7 @@ func Test_WrapWithEventsHandler(t *testing.T) { b, err := ioutil.ReadFile(c.AuditEvents.Name()) assert.NoError(err) - got := &eventJson{} + got := &cloudevents.Event{} err = json.Unmarshal(b, got) require.NoErrorf(err, "json: %s", string(b)) @@ -341,12 +342,13 @@ func Test_WrapWithEventsHandler(t *testing.T) { info := event.RequestInfo{ Method: "GET", Path: "/greeting", - Id: got.Payload["id"].(string), + // Id: got.Data.(map[string]interface{})["id"].(string), + Id: got.Data.(map[string]interface{})["request_info"].(map[string]interface{})["id"].(string), } hdr := map[string]interface{}{ - "id": got.Payload["id"].(string), - "timestamp": got.Payload["timestamp"].(string), - "serialized_hmac": got.Payload["serialized_hmac"].(string), + "id": got.Data.(map[string]interface{})["id"].(string), + "timestamp": got.Data.(map[string]interface{})["timestamp"].(string), + "serialized_hmac": got.Data.(map[string]interface{})["serialized_hmac"].(string), } wantJson := testJson(t, event.AuditType, &info, event.Op(tt.name), got, hdr, nil) assert.JSONEq(string(wantJson), string(actualJson)) @@ -397,7 +399,7 @@ func Test_startGatedEvents(t *testing.T) { Mutex: testLock, Name: "test", }) - e, err := event.NewEventer(testLogger, testLock, config, event.TestWithBroker(t, b)) + e, err := event.NewEventer(testLogger, testLock, tt.name, config, event.TestWithBroker(t, b)) require.NoError(err) ctx, err := event.NewEventerContext(context.Background(), e) require.NoError(err) @@ -456,7 +458,7 @@ func Test_flushGatedEvents(t *testing.T) { Mutex: testLock, Name: "test", }) - e, err := event.NewEventer(testLogger, testLock, config, event.TestWithBroker(t, b)) + e, err := event.NewEventer(testLogger, testLock, tt.name, config, event.TestWithBroker(t, b)) require.NoError(err) ctx, err := event.NewEventerContext(context.Background(), e) require.NoError(err) @@ -506,7 +508,7 @@ type eventJson struct { Payload map[string]interface{} `json:"payload"` } -func testJson(t *testing.T, eventType event.Type, reqInfo *event.RequestInfo, caller event.Op, got *eventJson, hdr, details map[string]interface{}) []byte { +func testJson(t *testing.T, eventType event.Type, reqInfo *event.RequestInfo, caller event.Op, got *cloudevents.Event, hdr, details map[string]interface{}) []byte { t.Helper() const ( testAuditVersion = "v0.1" @@ -520,19 +522,15 @@ func testJson(t *testing.T, eventType event.Type, reqInfo *event.RequestInfo, ca switch eventType { case event.ObservationType: payload = map[string]interface{}{ - event.IdField: got.Payload[event.IdField].(string), - event.HeaderField: map[string]interface{}{ - event.RequestInfoField: reqInfo, - event.VersionField: testObservationVersion, - }, + event.RequestInfoField: reqInfo, + event.VersionField: testObservationVersion, } - h := payload[event.HeaderField].(map[string]interface{}) for k, v := range hdr { - h[k] = v + payload[k] = v } case event.AuditType: payload = map[string]interface{}{ - event.IdField: got.Payload[event.IdField].(string), + event.IdField: got.Data.(map[string]interface{})[event.IdField].(string), event.RequestInfoField: reqInfo, event.VersionField: testAuditVersion, event.TypeField: event.ApiRequest, @@ -541,16 +539,20 @@ func testJson(t *testing.T, eventType event.Type, reqInfo *event.RequestInfo, ca payload[k] = v } } - j := eventJson{ - CreatedAt: got.CreatedAt, - EventType: string(eventType), - Payload: payload, + j := cloudevents.Event{ + ID: got.ID, + Time: got.Time, + Source: got.Source, + SpecVersion: got.SpecVersion, + Type: got.Type, + DataContentType: got.DataContentType, + Data: payload, } if details != nil { details[event.OpField] = string(caller) - d := got.Payload[event.DetailsField].([]interface{})[0].(map[string]interface{}) - j.Payload[event.DetailsField] = []struct { + d := got.Data.(map[string]interface{})[event.DetailsField].([]interface{})[0].(map[string]interface{}) + j.Data.(map[string]interface{})[event.DetailsField] = []struct { CreatedAt string `json:"created_at"` Type string `json:"type"` Payload map[string]interface{} `json:"payload"` diff --git a/internal/servers/controller/controller.go b/internal/servers/controller/controller.go index 5dc51e53d..29f00aeef 100644 --- a/internal/servers/controller/controller.go +++ b/internal/servers/controller/controller.go @@ -22,7 +22,6 @@ import ( "github.com/hashicorp/boundary/internal/session" "github.com/hashicorp/boundary/internal/target" "github.com/hashicorp/go-hclog" - "github.com/hashicorp/go-secure-stdlib/base62" "github.com/hashicorp/go-secure-stdlib/mlock" "github.com/patrickmn/go-cache" ua "go.uber.org/atomic" @@ -75,10 +74,9 @@ func New(conf *Config) (*Controller, error) { if conf.RawConfig.Controller == nil { conf.RawConfig.Controller = new(config.Controller) } - if conf.RawConfig.Controller.Name == "" { - if conf.RawConfig.Controller.Name, err = base62.Random(10); err != nil { - return nil, fmt.Errorf("error auto-generating controller name: %w", err) - } + + if conf.RawConfig.Controller.Name, err = conf.RawConfig.Controller.InitNameIfEmpty(); err != nil { + return nil, fmt.Errorf("error auto-generating controller name: %w", err) } if !conf.RawConfig.DisableMlock { diff --git a/internal/servers/controller/handlers/filtering.go b/internal/servers/controller/handlers/filtering.go index b0faa3556..dbca74446 100644 --- a/internal/servers/controller/handlers/filtering.go +++ b/internal/servers/controller/handlers/filtering.go @@ -1,13 +1,9 @@ package handlers import ( - "reflect" - "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/filter" "github.com/hashicorp/go-bexpr" - "google.golang.org/protobuf/types/known/structpb" - "google.golang.org/protobuf/types/known/timestamppb" - "google.golang.org/protobuf/types/known/wrapperspb" ) // filterItem captures all the different namespaces that can be used when @@ -27,7 +23,7 @@ func NewFilter(f string) (*Filter, error) { if f == "" { return &Filter{}, nil } - e, err := bexpr.CreateEvaluator(f, bexpr.WithTagName("json"), bexpr.WithHookFn(wellKnownTypeFilterHook)) + e, err := bexpr.CreateEvaluator(f, bexpr.WithTagName("json"), bexpr.WithHookFn(filter.WellKnownTypeFilterHook)) if err != nil { return nil, errors.Wrap(err, op, errors.WithMsg("couldn't build filter"), errors.WithCode(errors.InvalidParameter)) } @@ -49,37 +45,3 @@ func (f *Filter) Match(item interface{}) bool { // result in an error using the underlying library is simply interpreted as not a match. return err == nil && m } - -// wellKnownTypeFilterHook is passed to bexpr to treat all proto well-known -// types as the types they wrap for comparison. -func wellKnownTypeFilterHook(v reflect.Value) reflect.Value { - if !v.CanInterface() { - return v - } - ret := v.Interface() - switch pm := v.Interface().(type) { - case *wrapperspb.BoolValue: - ret = pm.GetValue() - case *wrapperspb.BytesValue: - ret = pm.GetValue() - case *wrapperspb.StringValue: - ret = pm.GetValue() - case *wrapperspb.DoubleValue: - ret = pm.GetValue() - case *wrapperspb.FloatValue: - ret = pm.GetValue() - case *wrapperspb.Int32Value: - ret = pm.GetValue() - case *wrapperspb.Int64Value: - ret = pm.GetValue() - case *wrapperspb.UInt32Value: - ret = pm.GetValue() - case *wrapperspb.UInt64Value: - ret = pm.GetValue() - case *structpb.Struct: - ret = pm.AsMap() - case *timestamppb.Timestamp: - ret = pm.AsTime() - } - return reflect.ValueOf(ret) -} diff --git a/internal/servers/controller/handlers/filtering_test.go b/internal/servers/controller/handlers/filtering_test.go index 7c26d857f..2882487f2 100644 --- a/internal/servers/controller/handlers/filtering_test.go +++ b/internal/servers/controller/handlers/filtering_test.go @@ -1,7 +1,6 @@ package handlers import ( - "reflect" "testing" "github.com/stretchr/testify/assert" @@ -118,20 +117,3 @@ func TestNewFilter(t *testing.T) { }) } } - -func TestWellKnownTypeFilterHook(t *testing.T) { - conversions := map[interface{}]interface{}{ - wrapperspb.String("foo"): "foo", - wrapperspb.UInt64(123): uint64(123), - wrapperspb.Int64(123): int64(123), - wrapperspb.UInt32(123): uint32(123), - wrapperspb.Int32(123): int32(123), - wrapperspb.Float(123): float32(123), - wrapperspb.Double(123): float64(123), - wrapperspb.Bytes([]byte("foo")): []byte("foo"), - wrapperspb.Bool(true): true, - } - for in, out := range conversions { - assert.Equal(t, reflect.ValueOf(out).Interface(), wellKnownTypeFilterHook(reflect.ValueOf(in)).Interface()) - } -} diff --git a/internal/servers/controller/testing.go b/internal/servers/controller/testing.go index ff4fd3b83..3e97c6f53 100644 --- a/internal/servers/controller/testing.go +++ b/internal/servers/controller/testing.go @@ -457,23 +457,24 @@ func NewTestController(t *testing.T, opts *TestControllerOpts) *TestController { }) } - if err := tc.b.SetupEventing(tc.b.Logger, tc.b.StderrLock, base.WithEventerConfig(opts.Config.Eventing)); err != nil { + if opts.Config.Controller == nil { + opts.Config.Controller = new(config.Controller) + } + if opts.Config.Controller.Name == "" { + opts.Config.Controller.Name, err = opts.Config.Controller.InitNameIfEmpty() + if err != nil { + t.Fatal(err) + } + tc.b.Logger.Info("controller name generated", "name", opts.Config.Controller.Name) + } + + if err := tc.b.SetupEventing(tc.b.Logger, tc.b.StderrLock, opts.Config.Controller.Name, base.WithEventerConfig(opts.Config.Eventing)); err != nil { t.Fatal(err) } // Initialize status grace period tc.b.SetStatusGracePeriodDuration(opts.StatusGracePeriodDuration) - if opts.Config.Controller == nil { - opts.Config.Controller = new(config.Controller) - } - if opts.Config.Controller.Name == "" { - opts.Config.Controller.Name, err = base62.Random(5) - if err != nil { - t.Fatal(err) - } - tc.b.Logger.Info("controller name generated", "name", opts.Config.Controller.Name) - } tc.name = opts.Config.Controller.Name if opts.InitialResourcesSuffix != "" { diff --git a/internal/servers/worker/testing.go b/internal/servers/worker/testing.go index 1e185f1b7..6801b5c40 100644 --- a/internal/servers/worker/testing.go +++ b/internal/servers/worker/testing.go @@ -230,7 +230,7 @@ func NewTestWorker(t *testing.T, opts *TestWorkerOpts) *TestWorker { opts.Config.Worker = new(config.Worker) } if opts.Config.Worker.Name == "" { - opts.Config.Worker.Name, err = base62.Random(5) + opts.Config.Worker.Name, err = opts.Config.Worker.InitNameIfEmpty() if err != nil { t.Fatal(err) } @@ -238,6 +238,10 @@ func NewTestWorker(t *testing.T, opts *TestWorkerOpts) *TestWorker { } tw.name = opts.Config.Worker.Name + if err := tw.b.SetupEventing(tw.b.Logger, tw.b.StderrLock, opts.Config.Worker.Name+"/test-worker", base.WithEventerConfig(opts.Config.Eventing)); err != nil { + t.Fatal(err) + } + // Set up KMSes switch { case opts.WorkerAuthKms != nil: diff --git a/internal/servers/worker/worker.go b/internal/servers/worker/worker.go index 57a39ea09..d4684f10a 100644 --- a/internal/servers/worker/worker.go +++ b/internal/servers/worker/worker.go @@ -12,7 +12,6 @@ import ( "github.com/hashicorp/boundary/internal/cmd/config" "github.com/hashicorp/boundary/internal/servers" "github.com/hashicorp/go-hclog" - "github.com/hashicorp/go-secure-stdlib/base62" "github.com/hashicorp/go-secure-stdlib/mlock" ua "go.uber.org/atomic" "google.golang.org/grpc/resolver" @@ -72,10 +71,8 @@ func New(conf *Config) (*Worker, error) { } var err error - if conf.RawConfig.Worker.Name == "" { - if conf.RawConfig.Worker.Name, err = base62.Random(10); err != nil { - return nil, fmt.Errorf("error auto-generating worker name: %w", err) - } + if conf.RawConfig.Worker.Name, err = w.conf.RawConfig.Worker.InitNameIfEmpty(); err != nil { + return nil, fmt.Errorf("error auto-generating worker name: %w", err) } if !conf.RawConfig.DisableMlock {