diff --git a/globals/boundary_developer_env.go b/globals/boundary_developer_env.go new file mode 100644 index 000000000..819216458 --- /dev/null +++ b/globals/boundary_developer_env.go @@ -0,0 +1,8 @@ +package globals + +const ( + // BOUNDARY_DEVELOPER_ENABLE_EVENTS defines an envvar that enables events. + // Once this feature is generally available, then this envvar will be + // deprecated and removed. + BOUNDARY_DEVELOPER_ENABLE_EVENTS = "BOUNDARY_DEVELOPER_ENABLE_EVENTS" +) diff --git a/internal/cmd/base/event_flags.go b/internal/cmd/base/event_flags.go new file mode 100644 index 000000000..5eb690758 --- /dev/null +++ b/internal/cmd/base/event_flags.go @@ -0,0 +1,21 @@ +package base + +import "github.com/hashicorp/boundary/internal/observability/event" + +// EventFlags represent the cmd flags supported overriding the configured or +// default event configuration +type EventFlags struct { + Format event.SinkFormat + AuditEnabled *bool + ObservationsEnabled *bool +} + +// Validate simply validates the flags +func (ef *EventFlags) Validate() error { + if ef != nil { + if err := ef.Format.Validate(); err != nil { + return err + } + } + return nil +} diff --git a/internal/cmd/base/event_flags_test.go b/internal/cmd/base/event_flags_test.go new file mode 100644 index 000000000..850e4d652 --- /dev/null +++ b/internal/cmd/base/event_flags_test.go @@ -0,0 +1,60 @@ +package base + +import ( + "testing" + + "github.com/hashicorp/boundary/internal/observability/event" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestEventFlags_Validate(t *testing.T) { + tests := []struct { + name string + flags EventFlags + wantErr bool + wantErrContains string + wantErrIs error + }{ + { + name: "valid-JsonFormat", + flags: EventFlags{ + Format: event.JSONSinkFormat, + }, + }, + { + name: "empty", + flags: EventFlags{}, + wantErr: true, + wantErrContains: "not a valid sink format", + wantErrIs: event.ErrInvalidParameter, + }, + { + name: "invalid-format", + flags: EventFlags{ + Format: "invalid-format", + }, + wantErr: true, + wantErrContains: "not a valid sink format", + wantErrIs: event.ErrInvalidParameter, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + err := tt.flags.Validate() + if tt.wantErr { + require.Error(err) + if tt.wantErrIs != nil { + assert.ErrorIs(err, tt.wantErrIs) + } + if tt.wantErrContains != "" { + assert.Contains(err.Error(), tt.wantErrContains) + } + return + } + require.NoError(err) + }) + } + +} diff --git a/internal/cmd/base/option.go b/internal/cmd/base/option.go index 5d725b0dd..748ba3d5c 100644 --- a/internal/cmd/base/option.go +++ b/internal/cmd/base/option.go @@ -1,10 +1,14 @@ package base +import "github.com/hashicorp/boundary/internal/observability/event" + // getOpts - iterate the inbound Options and return a struct. func getOpts(opt ...Option) Options { opts := getDefaultOptions() for _, o := range opt { - o(&opts) + if o != nil { + o(&opts) + } } return opts } @@ -24,6 +28,8 @@ type Options struct { withSkipTargetCreation bool withContainerImage string withDialect string + withEventerConfig *event.EventerConfig + withEventFlags *EventFlags withAttributeFieldPrefix string withStatusCode int } @@ -114,6 +120,20 @@ func withDialect(dialect string) Option { } } +// WithEventer allows an optional eventer config +func WithEventerConfig(config *event.EventerConfig) Option { + return func(o *Options) { + o.withEventerConfig = config + } +} + +// WithEventer allows an optional event configuration flags +func WithEventFlags(flags *EventFlags) Option { + return func(o *Options) { + o.withEventFlags = flags + } +} + // WithAttributeFieldPrefix tells the command what prefix // to attach to attribute fields when they are returned as errors. func WithAttributeFieldPrefix(p string) Option { diff --git a/internal/cmd/base/option_test.go b/internal/cmd/base/option_test.go new file mode 100644 index 000000000..269e83968 --- /dev/null +++ b/internal/cmd/base/option_test.go @@ -0,0 +1,119 @@ +package base + +import ( + "testing" + + "github.com/hashicorp/boundary/internal/observability/event" + "github.com/stretchr/testify/assert" +) + +// Test_GetOpts provides unit tests for GetOpts and all the options +func Test_GetOpts(t *testing.T) { + t.Parallel() + t.Run("nil-options", func(t *testing.T) { + assert := assert.New(t) + opts := getOpts(nil, nil) + testOpts := getDefaultOptions() + assert.Equal(opts, testOpts) + }) + t.Run("WithEventFlags", func(t *testing.T) { + assert := assert.New(t) + isTrue := true + f := EventFlags{ + Format: event.JSONSinkFormat, + AuditEnabled: &isTrue, + ObservationsEnabled: &isTrue, + } + opts := getOpts(WithEventFlags(&f)) + testOpts := getDefaultOptions() + testOpts.withEventFlags = &f + assert.Equal(opts, testOpts) + }) + t.Run("WithEventerConfig", func(t *testing.T) { + assert := assert.New(t) + c := event.EventerConfig{ + Sinks: []event.SinkConfig{ + // not a valid sink, but it doesn't need to be to test the + // option is properly supported. + { + Name: "test-sink", + SinkType: "Stderr", + }, + }, + } + opts := getOpts(WithEventerConfig(&c)) + testOpts := getDefaultOptions() + testOpts.withEventerConfig = &c + assert.Equal(opts, testOpts) + }) + t.Run("WithNoTokenScope", func(t *testing.T) { + assert := assert.New(t) + opts := getOpts(WithNoTokenScope()) + testOpts := getDefaultOptions() + testOpts.withNoTokenScope = true + assert.Equal(opts, testOpts) + }) + t.Run("WithSkipDatabaseDestruction", func(t *testing.T) { + assert := assert.New(t) + opts := getOpts(WithSkipDatabaseDestruction()) + testOpts := getDefaultOptions() + testOpts.withSkipDatabaseDestruction = true + assert.Equal(opts, testOpts) + }) + t.Run("WithNoTokenValue", func(t *testing.T) { + assert := assert.New(t) + opts := getOpts(WithNoTokenValue()) + testOpts := getDefaultOptions() + testOpts.withNoTokenValue = true + assert.Equal(opts, testOpts) + }) + t.Run("WithSkipAuthMethodCreation", func(t *testing.T) { + assert := assert.New(t) + opts := getOpts(WithSkipAuthMethodCreation()) + testOpts := getDefaultOptions() + testOpts.withSkipAuthMethodCreation = true + assert.Equal(opts, testOpts) + }) + t.Run("WithSkipOidcAuthMethodCreation", func(t *testing.T) { + assert := assert.New(t) + opts := getOpts(WithSkipOidcAuthMethodCreation()) + testOpts := getDefaultOptions() + testOpts.withSkipOidcAuthMethodCreation = true + assert.Equal(opts, testOpts) + }) + t.Run("WithSkipScopesCreation", func(t *testing.T) { + assert := assert.New(t) + opts := getOpts(WithSkipScopesCreation()) + testOpts := getDefaultOptions() + testOpts.withSkipScopesCreation = true + assert.Equal(opts, testOpts) + }) + t.Run("WithSkipHostResourcesCreation", func(t *testing.T) { + assert := assert.New(t) + opts := getOpts(WithSkipHostResourcesCreation()) + testOpts := getDefaultOptions() + testOpts.withSkipHostResourcesCreation = true + assert.Equal(opts, testOpts) + }) + t.Run("WithSkipTargetCreation", func(t *testing.T) { + assert := assert.New(t) + opts := getOpts(WithSkipTargetCreation()) + testOpts := getDefaultOptions() + testOpts.withSkipTargetCreation = true + assert.Equal(opts, testOpts) + }) + t.Run("WithContainerImage", func(t *testing.T) { + assert := assert.New(t) + opts := getOpts(WithContainerImage("test-container")) + testOpts := getDefaultOptions() + testOpts.withContainerImage = "test-container" + assert.Equal(opts, testOpts) + }) + t.Run("withDialect", func(t *testing.T) { + assert := assert.New(t) + opts := getOpts(withDialect("test-dialect")) + testOpts := getDefaultOptions() + testOpts.withDialect = "test-dialect" + assert.Equal(opts, testOpts) + }) +} diff --git a/internal/cmd/base/server_test.go b/internal/cmd/base/server_test.go new file mode 100644 index 000000000..ac641b5f4 --- /dev/null +++ b/internal/cmd/base/server_test.go @@ -0,0 +1,209 @@ +package base + +import ( + "context" + "sync" + "testing" + + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/observability/event" + "github.com/hashicorp/go-hclog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func Test_NewServer(t *testing.T) { + t.Run("assert-require-no-nil-fields", func(t *testing.T) { + assert := assert.New(t) + s := NewServer(&Command{}) + assert.Equal(s.Command, &Command{}) + assert.NotNil(s.InfoKeys) + assert.NotNil(s.Info) + assert.NotNil(s.SecureRandomReader) + assert.NotNil(s.ReloadFuncsLock) + assert.NotNil(s.ReloadFuncs) + assert.NotNil(s.StderrLock) + }) +} + +func TestServer_SetupEventing(t *testing.T) { + // DO NOT run these test in parallel since they have a dependency on + // event.sysEventer + + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + }) + setTrue := true + setFalse := false + + tests := []struct { + name string + s *Server + logger hclog.Logger + lock *sync.Mutex + opt []Option + want event.EventerConfig + wantErrMatch *errors.Template + wantErrIs error + wantErrContains string + }{ + { + name: "missing-logger", + s: &Server{}, + lock: testLock, + wantErrMatch: errors.T(errors.InvalidParameter), + wantErrContains: "missing logger", + }, + { + name: "missing-serialization-lock", + s: &Server{}, + logger: testLogger, + wantErrMatch: errors.T(errors.InvalidParameter), + wantErrContains: "missing serialization lock", + }, + { + name: "opts-none", + s: &Server{}, + logger: testLogger, + lock: testLock, + want: *event.DefaultEventerConfig(), + }, + { + name: "opts-event-flags", + s: &Server{}, + logger: testLogger, + lock: testLock, + opt: []Option{WithEventFlags(&EventFlags{ + Format: event.JSONSinkFormat, + AuditEnabled: &setTrue, + ObservationsEnabled: &setFalse, + })}, + want: func() event.EventerConfig { + c := event.DefaultEventerConfig() + c.AuditEnabled = true + c.ObservationsEnabled = false + return *c + }(), + }, + { + name: "opts-event-flags-invalid", + s: &Server{}, + logger: testLogger, + lock: testLock, + opt: []Option{WithEventFlags(&EventFlags{ + Format: "invalid-format", + })}, + wantErrIs: event.ErrInvalidParameter, + wantErrContains: "not a valid sink format", + }, + { + name: "opts-eventer-config", + s: &Server{}, + logger: testLogger, + lock: testLock, + opt: []Option{WithEventerConfig(&event.EventerConfig{ + ObservationsEnabled: false, + SysEventsEnabled: false, + AuditEnabled: true, + })}, + want: func() event.EventerConfig { + c := event.DefaultEventerConfig() + c.AuditEnabled = true + c.ObservationsEnabled = false + c.SysEventsEnabled = false + return *c + }(), + }, + { + name: "opts-eventer-config-invalid", + s: &Server{}, + logger: testLogger, + lock: testLock, + opt: []Option{WithEventerConfig(&event.EventerConfig{ + Sinks: []event.SinkConfig{ + { + Format: "invalid-format", + }, + }, + })}, + wantErrIs: event.ErrInvalidParameter, + wantErrContains: "sink 0 is invalid", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + event.TestResetSystEventer(t) + + err := tt.s.SetupEventing(tt.logger, tt.lock, tt.opt...) + if tt.wantErrMatch != nil || tt.wantErrIs != nil { + require.Error(err) + assert.Nil(tt.s.Eventer) + assert.Nil(event.SysEventer()) + if tt.wantErrMatch != nil { + assert.Truef(errors.Match(tt.wantErrMatch, err), "want %q and got %q", tt.wantErrMatch.Code, err.Error()) + } + if tt.wantErrIs != nil { + assert.ErrorIs(err, tt.wantErrIs) + } + if tt.wantErrContains != "" { + assert.Contains(err.Error(), tt.wantErrContains) + } + return + } + require.NoError(err) + assert.Equal(tt.want, event.TestGetEventerConfig(t, tt.s.Eventer)) + }) + } +} + +func TestServer_AddEventerToContext(t *testing.T) { + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + }) + testEventer, err := event.NewEventer(testLogger, testLock, event.EventerConfig{}) + require.NoError(t, err) + tests := []struct { + name string + s Server + ctx context.Context + wantErrMatch *errors.Template + wantErrContains string + }{ + { + name: "missing-eventer", + s: Server{}, + ctx: context.Background(), + wantErrMatch: errors.T(errors.InvalidParameter), + wantErrContains: "missing server eventer", + }, + { + name: "valid", + s: Server{Eventer: testEventer}, + ctx: context.Background(), + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + gotCtx, err := tt.s.AddEventerToContext(tt.ctx) + if tt.wantErrMatch != nil { + require.Error(err) + assert.Nil(gotCtx) + assert.Truef(errors.Match(tt.wantErrMatch, err), "want %q and got %q", tt.wantErrMatch.Code, err.Error()) + if tt.wantErrContains != "" { + assert.Contains(err.Error(), tt.wantErrContains) + } + return + } + require.NoError(err) + e, ok := event.EventerFromContext(gotCtx) + require.Truef(ok, "unable to get eventer from context") + assert.NotNil(e) + assert.Equal(tt.s.Eventer, e) + }) + } +} diff --git a/internal/cmd/base/servers.go b/internal/cmd/base/servers.go index 6088fa620..aacd73716 100644 --- a/internal/cmd/base/servers.go +++ b/internal/cmd/base/servers.go @@ -17,10 +17,13 @@ import ( "syscall" "github.com/armon/go-metrics" + berrors "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/globals" "github.com/hashicorp/boundary/internal/cmd/config" "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/kms" + "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/boundary/internal/types/scope" "github.com/hashicorp/boundary/sdk/strutil" "github.com/hashicorp/boundary/version" @@ -51,6 +54,9 @@ type Server struct { CombineLogs bool LogLevel hclog.Level + StderrLock *sync.Mutex + Eventer *event.Eventer + RootKms wrapping.Wrapper WorkerAuthKms wrapping.Wrapper RecoveryKms wrapping.Wrapper @@ -107,9 +113,75 @@ func NewServer(cmd *Command) *Server { SecureRandomReader: rand.Reader, ReloadFuncsLock: new(sync.RWMutex), ReloadFuncs: make(map[string][]reloadutil.ReloadFunc), + StderrLock: new(sync.Mutex), } } +// 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 { + const op = "base.(Server).SetupEventing" + + if logger == nil { + return berrors.New(berrors.InvalidParameter, op, "missing logger") + } + if serializationLock == nil { + return berrors.New(berrors.InvalidParameter, op, "missing serialization lock") + } + + opts := getOpts(opt...) + if opts.withEventerConfig != nil { + if err := opts.withEventerConfig.Validate(); err != nil { + return berrors.Wrap(err, op, berrors.WithMsg("invalid eventer config")) + } + } + if opts.withEventerConfig == nil { + opts.withEventerConfig = event.DefaultEventerConfig() + } + + if opts.withEventFlags != nil { + if err := opts.withEventFlags.Validate(); err != nil { + 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 + } + } + if opts.withEventFlags.AuditEnabled != nil { + opts.withEventerConfig.AuditEnabled = *opts.withEventFlags.AuditEnabled + } + if opts.withEventFlags.ObservationsEnabled != nil { + opts.withEventerConfig.ObservationsEnabled = *opts.withEventFlags.ObservationsEnabled + } + } + + e, err := event.NewEventer(logger, serializationLock, *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 { + return berrors.Wrap(err, op, berrors.WithMsg("unable to initialize system eventer")) + } + + return nil +} + +// AddEventerToContext will add the server eventer to the context provided +func (b *Server) AddEventerToContext(ctx context.Context) (context.Context, error) { + const op = "base.(Server).AddEventerToContext" + if b.Eventer == nil { + return nil, berrors.New(berrors.InvalidParameter, op, "missing server eventer") + } + e, err := event.NewEventerContext(ctx, b.Eventer) + if err != nil { + return nil, berrors.Wrap(err, op, berrors.WithMsg("unable to add eventer to context")) + } + return e, nil +} + func (b *Server) SetupLogging(flagLogLevel, flagLogFormat, configLogLevel, configLogFormat string) error { b.logOutput = os.Stderr if b.CombineLogs { @@ -128,6 +200,7 @@ func (b *Server) SetupLogging(flagLogLevel, flagLogFormat, configLogLevel, confi // Note that if logFormat is either unspecified or standard, then // the resulting logger's format will be standard. JSONFormat: logFormat == logging.JSONFormat, + Mutex: b.StderrLock, }) // create GRPC logger diff --git a/internal/cmd/commands/database/init.go b/internal/cmd/commands/database/init.go index 85fa0db06..69ad70fb1 100644 --- a/internal/cmd/commands/database/init.go +++ b/internal/cmd/commands/database/init.go @@ -185,6 +185,11 @@ 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 { + c.UI.Error(err.Error()) + return base.CommandCliError + } + if err := c.srv.SetupKMSes(c.UI, c.Config); 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 092e627b4..2d04dab76 100644 --- a/internal/cmd/commands/database/migrate.go +++ b/internal/cmd/commands/database/migrate.go @@ -137,6 +137,11 @@ func (c *MigrateCommand) 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 { + c.UI.Error(err.Error()) + return base.CommandCliError + } + // If mlockall(2) isn't supported, show a warning. We disable this in dev // because it is quite scary to see when first using Boundary. We also disable // this if the user has explicitly disabled mlock in configuration. diff --git a/internal/cmd/commands/dev/dev.go b/internal/cmd/commands/dev/dev.go index 98d63c559..955b5de1d 100644 --- a/internal/cmd/commands/dev/dev.go +++ b/internal/cmd/commands/dev/dev.go @@ -420,6 +420,11 @@ 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 { + c.UI.Error(err.Error()) + return base.CommandCliError + } + base.StartMemProfiler(c.Logger) if err := c.SetupMetrics(c.UI, c.Config.Telemetry); err != nil { diff --git a/internal/cmd/commands/server/server.go b/internal/cmd/commands/server/server.go index 3ad86c833..880accdb8 100644 --- a/internal/cmd/commands/server/server.go +++ b/internal/cmd/commands/server/server.go @@ -151,6 +151,11 @@ 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 { + c.UI.Error(err.Error()) + return base.CommandUserError + } + base.StartMemProfiler(c.Logger) if !c.skipMetrics { diff --git a/internal/cmd/commands/server/server_test.go b/internal/cmd/commands/server/server_test.go index 0f6f23a51..939e51290 100644 --- a/internal/cmd/commands/server/server_test.go +++ b/internal/cmd/commands/server/server_test.go @@ -30,6 +30,7 @@ func testServerCommand(t *testing.T, controllerKey string) *Command { } require.NoError(cmd.SetupLogging("trace", "", "", "")) + require.NoError(cmd.SetupEventing(cmd.Logger, cmd.StderrLock)) 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 679a940b9..0cf569839 100644 --- a/internal/cmd/config/config.go +++ b/internal/cmd/config/config.go @@ -11,6 +11,7 @@ import ( "strings" "time" + "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/boundary/sdk/strutil" wrapping "github.com/hashicorp/go-kms-wrapping" "github.com/hashicorp/hcl" @@ -99,6 +100,9 @@ type Config struct { DevControllerKey string `hcl:"-"` DevWorkerAuthKey string `hcl:"-"` DevRecoveryKey string `hcl:"-"` + + // Eventing configuration for the controller + Eventing *event.EventerConfig `hcl:"events"` } type Controller struct { @@ -359,6 +363,14 @@ func Parse(d string) (*Config, error) { } } + if result.Eventing == nil { + result.Eventing = event.DefaultEventerConfig() + } else { + if result.Eventing.Sinks == nil { + result.Eventing.Sinks = []event.SinkConfig{event.DefaultSink()} + } + } + return result, nil } diff --git a/internal/cmd/config/config_test.go b/internal/cmd/config/config_test.go index c1adc7927..dd75e61c1 100644 --- a/internal/cmd/config/config_test.go +++ b/internal/cmd/config/config_test.go @@ -6,6 +6,8 @@ import ( "testing" "time" + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/shared-secure-libs/configutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -20,6 +22,7 @@ func TestDevController(t *testing.T) { truePointer := new(bool) *truePointer = true exp := &Config{ + Eventing: event.DefaultEventerConfig(), SharedConfig: &configutil.SharedConfig{ DisableMlock: true, Listeners: []*configutil.Listener{ @@ -152,6 +155,7 @@ func TestDevWorker(t *testing.T) { } exp := &Config{ + Eventing: event.DefaultEventerConfig(), SharedConfig: &configutil.SharedConfig{ DisableMlock: true, Listeners: []*configutil.Listener{ @@ -331,3 +335,92 @@ func TestParsingName(t *testing.T) { }) } } + +func TestController_EventingConfig(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + config string + wantEventerConfig *event.EventerConfig + wantErrMatch *errors.Template + }{ + { + name: "default", + wantEventerConfig: event.DefaultEventerConfig(), + }, + { + name: "audit-enabled", + config: ` + events { + audit_enabled = true + } + `, + wantEventerConfig: &event.EventerConfig{ + AuditEnabled: true, + ObservationsEnabled: false, + Sinks: []event.SinkConfig{ + event.DefaultSink(), + }, + }, + }, + { + name: "observations-enabled", + config: ` + events { + observations_enabled = true + } + `, + wantEventerConfig: &event.EventerConfig{ + AuditEnabled: false, + ObservationsEnabled: true, + Sinks: []event.SinkConfig{ + event.DefaultSink(), + }, + }, + }, + { + name: "sinks-configured", + config: ` + events { + audit_enabled = false + observations_enabled = true + sinks = [ + { + name = "configured-sink" + file_name = "file-name" + event_types = [ "audit", "observation" ] + } + ] + } + `, + wantEventerConfig: &event.EventerConfig{ + AuditEnabled: false, + ObservationsEnabled: true, + Sinks: []event.SinkConfig{ + { + Name: "configured-sink", + FileName: "file-name", + EventTypes: []event.Type{"audit", "observation"}, + }, + }, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + c, err := Parse(tt.config) + if tt.wantErrMatch != nil { + require.NoError(err) + assert.Empty(c) + assert.Truef(errors.Match(tt.wantErrMatch, err), "want %q and got %q", tt.wantErrMatch.Code, err.Error()) + return + } + require.NoError(err) + assert.NotEmpty(c) + assert.Equal(tt.wantEventerConfig, c.Eventing) + }) + + } +} diff --git a/internal/observability/event/context.go b/internal/observability/event/context.go index 64c0becd5..b2969f143 100644 --- a/internal/observability/event/context.go +++ b/internal/observability/event/context.go @@ -3,8 +3,11 @@ package event import ( "context" "fmt" + "os" "runtime" + "strings" + "github.com/hashicorp/boundary/globals" "github.com/hashicorp/go-hclog" ) @@ -70,6 +73,11 @@ func RequestInfoFromContext(ctx context.Context) (*RequestInfo, bool) { // WithHeader, WithDetails, WithId, WithFlush and WithRequestInfo. All other // options are ignored. func WriteObservation(ctx context.Context, caller Op, opt ...Option) error { + // TODO (jimlambrt) 6/2021: remove this feature flag envvar when events are + // generally available. + if !strings.EqualFold(os.Getenv(globals.BOUNDARY_DEVELOPER_ENABLE_EVENTS), "true") { + return nil + } const op = "event.WriteObservation" if ctx == nil { return fmt.Errorf("%s: missing context: %w", op, ErrInvalidParameter) @@ -111,6 +119,11 @@ func WriteObservation(ctx context.Context, caller Op, opt ...Option) error { // The options WithId and WithRequestInfo are supported and all other options // are ignored. func WriteError(ctx context.Context, caller Op, e error, opt ...Option) { + // TODO (jimlambrt) 6/2021: remove this feature flag envvar when events are + // generally available. + if !strings.EqualFold(os.Getenv(globals.BOUNDARY_DEVELOPER_ENABLE_EVENTS), "true") { + return + } const op = "event.WriteError" // EventerFromContext will handle a nil ctx appropriately. If e or caller is // missing, newError(...) will handle them appropriately. @@ -153,6 +166,11 @@ func WriteError(ctx context.Context, caller Op, e error, opt ...Option) { // WithRequest, WithResponse, WithAuth, WithId, WithFlush and WithRequestInfo. // All other options are ignored. func WriteAudit(ctx context.Context, caller Op, opt ...Option) error { + // TODO (jimlambrt) 6/2021: remove this feature flag envvar when events are + // generally available. + if !strings.EqualFold(os.Getenv(globals.BOUNDARY_DEVELOPER_ENABLE_EVENTS), "true") { + return nil + } const op = "event.WriteAudit" if ctx == nil { return fmt.Errorf("%s: missing context: %w", op, ErrInvalidParameter) diff --git a/internal/observability/event/context_test.go b/internal/observability/event/context_test.go index f8bdd8cf6..a3804c48a 100644 --- a/internal/observability/event/context_test.go +++ b/internal/observability/event/context_test.go @@ -6,6 +6,7 @@ import ( "fmt" "io/ioutil" "os" + "sync" "testing" "time" @@ -146,7 +147,12 @@ func Test_RequestInfoFromContext(t *testing.T) { func Test_NewEventerContext(t *testing.T) { testSetup := event.TestEventerConfig(t, "Test_NewEventerContext") - testEventer, err := event.NewEventer(hclog.Default(), testSetup.EventerConfig) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + testEventer, err := event.NewEventer(testLogger, testLock, testSetup.EventerConfig) require.NoError(t, err) tests := []struct { name string @@ -198,7 +204,12 @@ func Test_NewEventerContext(t *testing.T) { func Test_EventerFromContext(t *testing.T) { testSetup := event.TestEventerConfig(t, "Test_EventerFromContext") - testEventer, err := event.NewEventer(hclog.Default(), testSetup.EventerConfig) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + testEventer, err := event.NewEventer(testLogger, testLock, testSetup.EventerConfig) require.NoError(t, err) testEventerCtx, err := event.NewEventerContext(context.Background(), testEventer) @@ -242,15 +253,19 @@ func Test_EventerFromContext(t *testing.T) { } func Test_WriteObservation(t *testing.T) { + event.TestEnableEventing(t, true) + // this test and its subtests cannot be run in parallel because of it's // dependency on the sysEventer - logger := hclog.New(&hclog.LoggerOptions{ - Name: "test", - }) c := event.TestEventerConfig(t, "WriteObservation") - e, err := event.NewEventer(logger, c.EventerConfig) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + e, err := event.NewEventer(testLogger, testLock, c.EventerConfig) require.NoError(t, err) info := &event.RequestInfo{Id: "867-5309"} @@ -329,7 +344,7 @@ func Test_WriteObservation(t *testing.T) { }, observationSinkFileName: c.AllEvents.Name(), setup: func() error { - return event.InitSysEventer(hclog.Default(), c.EventerConfig) + return event.InitSysEventer(testLogger, testLock, event.WithEventerConfig(&c.EventerConfig)) }, cleanup: func() { event.TestResetSystEventer(t) }, }, @@ -380,7 +395,7 @@ func Test_WriteObservation(t *testing.T) { }, observationSinkFileName: c.AllEvents.Name(), setup: func() error { - return event.InitSysEventer(hclog.Default(), c.EventerConfig) + return event.InitSysEventer(testLogger, testLock, event.WithEventerConfig(&c.EventerConfig)) }, cleanup: func() { event.TestResetSystEventer(t) }, }, @@ -441,14 +456,15 @@ func Test_WriteObservation(t *testing.T) { } t.Run("not-enabled", func(t *testing.T) { assert, require := assert.New(t), require.New(t) - logger := hclog.New(&hclog.LoggerOptions{ - Name: "test", - }) c := event.TestEventerConfig(t, "WriteObservation") c.EventerConfig.ObservationsEnabled = false - - e, err := event.NewEventer(logger, c.EventerConfig) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + e, err := event.NewEventer(testLogger, testLock, c.EventerConfig) require.NoError(err) testCtx, err := event.NewEventerContext(context.Background(), e) @@ -518,17 +534,19 @@ type eventJson struct { } func Test_WriteAudit(t *testing.T) { + event.TestEnableEventing(t, true) + // this test and its subtests cannot be run in parallel because of it's // dependency on the sysEventer now := time.Now() - logger := hclog.New(&hclog.LoggerOptions{ - Name: "test", - }) - c := event.TestEventerConfig(t, "WriteAudit") - - e, err := event.NewEventer(logger, c.EventerConfig, event.WithNow(now)) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + e, err := event.NewEventer(testLogger, testLock, c.EventerConfig) require.NoError(t, err) info := &event.RequestInfo{Id: "867-5309"} @@ -635,7 +653,7 @@ func Test_WriteAudit(t *testing.T) { Request: testReq, }, setup: func() error { - return event.InitSysEventer(hclog.Default(), c.EventerConfig) + return event.InitSysEventer(testLogger, testLock, event.WithEventerConfig(&c.EventerConfig)) }, cleanup: func() { event.TestResetSystEventer(t) }, auditSinkFileName: c.AllEvents.Name(), @@ -733,14 +751,14 @@ func Test_WriteAudit(t *testing.T) { } t.Run("not-enabled", func(t *testing.T) { assert, require := assert.New(t), require.New(t) - logger := hclog.New(&hclog.LoggerOptions{ - Name: "test", - }) - c := event.TestEventerConfig(t, "WriteAudit") c.EventerConfig.AuditEnabled = false - - e, err := event.NewEventer(logger, c.EventerConfig) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + e, err := event.NewEventer(testLogger, testLock, c.EventerConfig) require.NoError(err) testCtx, err := event.NewEventerContext(context.Background(), e) @@ -756,17 +774,19 @@ func Test_WriteAudit(t *testing.T) { } func Test_WriteError(t *testing.T) { + event.TestEnableEventing(t, true) + // this test and its subtests cannot be run in parallel because of it's // dependency on the sysEventer now := time.Now() - logger := hclog.New(&hclog.LoggerOptions{ - Name: "test", - }) - c := event.TestEventerConfig(t, "WriteAudit") - - e, err := event.NewEventer(logger, c.EventerConfig, event.WithNow(now)) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + e, err := event.NewEventer(testLogger, testLock, c.EventerConfig, event.WithNow(now)) require.NoError(t, err) info := &event.RequestInfo{Id: "867-5309"} @@ -818,7 +838,7 @@ func Test_WriteError(t *testing.T) { ctx: context.Background(), e: &testError, setup: func() error { - return event.InitSysEventer(hclog.Default(), c.EventerConfig) + return event.InitSysEventer(testLogger, testLock, event.WithEventerConfig(&c.EventerConfig)) }, cleanup: func() { event.TestResetSystEventer(t) }, errSinkFileName: c.ErrorEvents.Name(), @@ -829,7 +849,7 @@ func Test_WriteError(t *testing.T) { e: &testError, info: &event.RequestInfo{}, setup: func() error { - return event.InitSysEventer(hclog.Default(), c.EventerConfig) + return event.InitSysEventer(testLogger, testLock, event.WithEventerConfig(&c.EventerConfig)) }, cleanup: func() { event.TestResetSystEventer(t) }, errSinkFileName: c.ErrorEvents.Name(), @@ -899,6 +919,11 @@ func Test_WriteSysEvent(t *testing.T) { ctx := context.Background() c := event.TestEventerConfig(t, "Test_WriteSysEvent") + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) tests := []struct { name string @@ -934,7 +959,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(hclog.Default(), c.EventerConfig) + return event.InitSysEventer(testLogger, testLock, event.WithEventerConfig(&c.EventerConfig)) }, cleanup: func() { event.TestResetSystEventer(t) }, sinkFileName: c.AllEvents.Name(), diff --git a/internal/observability/event/event_audit.go b/internal/observability/event/event_audit.go index 3f324ea35..1c805d5f2 100644 --- a/internal/observability/event/event_audit.go +++ b/internal/observability/event/event_audit.go @@ -14,7 +14,7 @@ const auditVersion = "v0.1" type auditEventType string const ( - apiRequest auditEventType = "APIRequest" // ApiRequest defines an API request audit event type + ApiRequest auditEventType = "APIRequest" // ApiRequest defines an API request audit event type ) // audit defines the data of audit events @@ -55,7 +55,7 @@ func newAudit(fromOperation Op, opt ...Option) (*audit, error) { a := &audit{ Id: opts.withId, Version: auditVersion, - Type: string(apiRequest), + Type: string(ApiRequest), Timestamp: dtm, RequestInfo: opts.withRequestInfo, Auth: opts.withAuth, @@ -120,7 +120,7 @@ func (a *audit) ComposeFrom(events []*eventlogger.Event) (eventlogger.EventType, if gated.Version != auditVersion { return "", nil, fmt.Errorf("%s: event %d has an invalid version: %s != %s: %w", op, i, gated.Version, auditVersion, ErrInvalidParameter) } - if gated.Type != string(apiRequest) { + if gated.Type != string(ApiRequest) { return "", nil, fmt.Errorf("%s: event %d has an invalid type: %s != %s: %w", op, i, gated.Type, string(AuditType), ErrInvalidParameter) } if gated.RequestInfo != nil { @@ -142,6 +142,6 @@ func (a *audit) ComposeFrom(events []*eventlogger.Event) (eventlogger.EventType, } payload.Id = validId payload.Version = auditVersion - payload.Type = string(apiRequest) + payload.Type = string(ApiRequest) return eventlogger.EventType(a.EventType()), payload, nil } diff --git a/internal/observability/event/event_audit_test.go b/internal/observability/event/event_audit_test.go index ae9fe5523..37de0d5c4 100644 --- a/internal/observability/event/event_audit_test.go +++ b/internal/observability/event/event_audit_test.go @@ -29,7 +29,7 @@ func Test_newAudit(t *testing.T) { fromOp: "valid-no-opts", want: &audit{ Version: auditVersion, - Type: string(apiRequest), + Type: string(ApiRequest), }, }, { @@ -47,7 +47,7 @@ func Test_newAudit(t *testing.T) { want: &audit{ Id: "all-opts", Version: auditVersion, - Type: string(apiRequest), + Type: string(ApiRequest), Timestamp: testNow, RequestInfo: TestRequestInfo(t), Auth: testAuth(t), @@ -181,7 +181,7 @@ func TestAudit_ComposeFrom(t *testing.T) { Payload: &audit{ Id: "test-id", Version: "invalid-version", - Type: string(apiRequest), + Type: string(ApiRequest), }, }, }, @@ -195,14 +195,14 @@ func TestAudit_ComposeFrom(t *testing.T) { Payload: &audit{ Id: "invalid-id", Version: auditVersion, - Type: string(apiRequest), + Type: string(ApiRequest), }, }, { Payload: &audit{ Id: "bad-id", Version: auditVersion, - Type: string(apiRequest), + Type: string(ApiRequest), }, }, }, @@ -216,7 +216,7 @@ func TestAudit_ComposeFrom(t *testing.T) { Payload: &audit{ Id: "valid", Version: auditVersion, - Type: string(apiRequest), + Type: string(ApiRequest), Timestamp: testNow, Auth: testAuth(t), RequestInfo: TestRequestInfo(t), @@ -226,7 +226,7 @@ func TestAudit_ComposeFrom(t *testing.T) { Payload: &audit{ Id: "valid", Version: auditVersion, - Type: string(apiRequest), + Type: string(ApiRequest), Timestamp: testNow, Request: testRequest(t), }, @@ -235,7 +235,7 @@ func TestAudit_ComposeFrom(t *testing.T) { Payload: &audit{ Id: "valid", Version: auditVersion, - Type: string(apiRequest), + Type: string(ApiRequest), Timestamp: testNow, Response: testResponse(t), }, @@ -244,7 +244,7 @@ func TestAudit_ComposeFrom(t *testing.T) { want: audit{ Id: "valid", Version: auditVersion, - Type: string(apiRequest), + Type: string(ApiRequest), Timestamp: testNow, Auth: testAuth(t), Request: testRequest(t), diff --git a/internal/observability/event/eventer.go b/internal/observability/event/eventer.go index 48e82f4b8..64b0c205b 100644 --- a/internal/observability/event/eventer.go +++ b/internal/observability/event/eventer.go @@ -47,69 +47,101 @@ type broker interface { // Eventer provides a method to send events to pipelines of sinks type Eventer struct { - broker broker - flushableNodes []flushable - conf EventerConfig - logger hclog.Logger + broker broker + flushableNodes []flushable + conf EventerConfig + logger hclog.Logger + auditPipelines []pipeline + observationPipelines []pipeline + errPipelines []pipeline +} + +type pipeline struct { + eventType Type + fmtId eventlogger.NodeID + sinkId eventlogger.NodeID + gateId eventlogger.NodeID + sinkConfig SinkConfig } var ( - sysEventer *Eventer // sysEventer is the system-wide Eventer - sysEventerOnce sync.Once // sysEventerOnce ensures that the system-wide Eventer is only initialized once. + sysEventer *Eventer // sysEventer is the system-wide Eventer + sysEventerLock sync.RWMutex // sysEventerLock allows the sysEventer to safely be written concurrently. ) // InitSysEventer provides a mechanism to initialize a "system wide" eventer -// singleton for Boundary -func InitSysEventer(log hclog.Logger, c EventerConfig) error { +// singleton for Boundary. Support the options of: WithEventer(...) and +// WithEventerConfig(...) +// +// IMPORTANT: Eventers cannot share file sinks, which likely means that each +// 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 { const op = "event.InitSysEventer" if log == nil { return fmt.Errorf("%s: missing hclog: %w", op, ErrInvalidParameter) } - // the order of operations is important here. we want to determine if - // there's an error before setting the singleton sysEventer which can only - // be done one time. - eventer, err := NewEventer(log, c) - if err != nil { - return fmt.Errorf("%s: %w", op, err) + if serializationLock == nil { + return fmt.Errorf("%s: missing serialization lock: %w", op, ErrInvalidParameter) } - sysEventerOnce.Do(func() { - sysEventer = eventer - }) + + // the order of operations is important here. we want to determine if + // there's an error before setting the singleton sysEventer + var e *Eventer + opts := getOpts(opt...) + switch { + case opts.withEventer == nil && opts.withEventerConfig == nil: + return fmt.Errorf("%s: missing both eventer and eventer config: %w", op, ErrInvalidParameter) + + case opts.withEventer != nil && opts.withEventerConfig != nil: + return fmt.Errorf("%s: both eventer and eventer config provided: %w", op, ErrInvalidParameter) + + case opts.withEventerConfig != nil: + var err error + if e, err = NewEventer(log, serializationLock, *opts.withEventerConfig); err != nil { + return fmt.Errorf("%s: %w", op, err) + } + + case opts.withEventer != nil: + e = opts.withEventer + } + + sysEventerLock.Lock() + defer sysEventerLock.Unlock() + sysEventer = e return nil } -// SysEventer returns the "system wide" eventer for Boundary. +// SysEventer returns the "system wide" eventer for Boundary and can/will return +// a nil Eventer func SysEventer() *Eventer { + sysEventerLock.RLock() + defer sysEventerLock.RUnlock() return sysEventer } -// NewEventer creates a new Eventer using the config. Supports options: WithNow -func NewEventer(log hclog.Logger, c EventerConfig, opt ...Option) (*Eventer, error) { +// 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) { const op = "event.NewEventer" if log == nil { return nil, fmt.Errorf("%s: missing logger: %w", op, ErrInvalidParameter) } - - // if there are no sinks in config, then we'll default to just one stdout - // sink. - if len(c.Sinks) == 0 { - c.Sinks = append(c.Sinks, SinkConfig{ - Name: "default", - EventTypes: []Type{EveryType}, - Format: JSONSinkFormat, - SinkType: StdoutSink, - }) + if serializationLock == nil { + return nil, fmt.Errorf("%s: missing serialization lock: %w", op, ErrInvalidParameter) } - if err := c.validate(); err != nil { + // if there are no sinks in config, then we'll default to just one stderr + // sink. + if len(c.Sinks) == 0 { + c.Sinks = append(c.Sinks, DefaultSink()) + } + + if err := c.Validate(); err != nil { return nil, fmt.Errorf("%s: %w", op, err) } - type pipeline struct { - eventType Type - fmtId eventlogger.NodeID - sinkId eventlogger.NodeID - } var auditPipelines, observationPipelines, errPipelines, sysPipelines []pipeline opts := getOpts(opt...) @@ -143,21 +175,35 @@ func NewEventer(log hclog.Logger, c EventerConfig, opt ...Option) (*Eventer, err 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{ + w: os.Stderr, + l: serializationLock, + } + + // we need to keep track of all the Sink filenames to ensure they aren't + // reused. + allSinkFilenames := map[string]bool{} + for _, s := range c.Sinks { var sinkId eventlogger.NodeID var sinkNode eventlogger.Node switch s.SinkType { - case StdoutSink: + case StderrSink: sinkNode = &eventlogger.WriterSink{ Format: string(s.Format), - Writer: os.Stdout, + Writer: &serializedStderr, } - id, err = newId("stdout") + id, err = newId("stderr") if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } sinkId = eventlogger.NodeID(id) default: + if _, found := allSinkFilenames[s.Path+s.FileName]; found { + return nil, fmt.Errorf("%s: duplicate file sink: %s %s", op, s.Path, s.FileName) + } sinkNode = &eventlogger.FileSink{ Format: string(s.Format), Path: s.Path, @@ -196,23 +242,26 @@ func NewEventer(log hclog.Logger, c EventerConfig, opt ...Option) (*Eventer, err } if addToAudit { auditPipelines = append(auditPipelines, pipeline{ - eventType: AuditType, - fmtId: jsonfmtId, - sinkId: sinkId, + eventType: AuditType, + fmtId: jsonfmtId, + sinkId: sinkId, + sinkConfig: s, }) } if addToObservation { observationPipelines = append(observationPipelines, pipeline{ - eventType: ObservationType, - fmtId: jsonfmtId, - sinkId: sinkId, + eventType: ObservationType, + fmtId: jsonfmtId, + sinkId: sinkId, + sinkConfig: s, }) } if addToErr { errPipelines = append(errPipelines, pipeline{ - eventType: ErrorType, - fmtId: jsonfmtId, - sinkId: sinkId, + eventType: ErrorType, + fmtId: jsonfmtId, + sinkId: sinkId, + sinkConfig: s, }) } if addToSys { @@ -233,7 +282,6 @@ func NewEventer(log hclog.Logger, c EventerConfig, opt ...Option) (*Eventer, err 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 := eventlogger.GatedFilter{} e.flushableNodes = append(e.flushableNodes, &gatedFilterNode) @@ -241,8 +289,8 @@ func NewEventer(log hclog.Logger, c EventerConfig, opt ...Option) (*Eventer, err if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } - gatedFilterNodeId := eventlogger.NodeID(gateId) - if err := e.broker.RegisterNode(gatedFilterNodeId, &gatedFilterNode); err != nil { + p.gateId = eventlogger.NodeID(gateId) + if err := e.broker.RegisterNode(p.gateId, &gatedFilterNode); err != nil { return nil, fmt.Errorf("%s: unable to register audit gated filter: %w", op, err) } @@ -253,14 +301,13 @@ func NewEventer(log hclog.Logger, c EventerConfig, opt ...Option) (*Eventer, err err = e.broker.RegisterPipeline(eventlogger.Pipeline{ EventType: eventlogger.EventType(p.eventType), PipelineID: eventlogger.PipelineID(pipeId), - NodeIDs: []eventlogger.NodeID{gatedFilterNodeId, p.fmtId, p.sinkId}, + NodeIDs: []eventlogger.NodeID{p.gateId, p.fmtId, p.sinkId}, }) 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 := eventlogger.GatedFilter{} e.flushableNodes = append(e.flushableNodes, &gatedFilterNode) @@ -268,8 +315,8 @@ func NewEventer(log hclog.Logger, c EventerConfig, opt ...Option) (*Eventer, err if err != nil { return nil, fmt.Errorf("%s: %w", op, err) } - gatedFilterNodeId := eventlogger.NodeID(gateId) - if err := e.broker.RegisterNode(gatedFilterNodeId, &gatedFilterNode); err != nil { + p.gateId = eventlogger.NodeID(gateId) + if err := e.broker.RegisterNode(p.gateId, &gatedFilterNode); err != nil { return nil, fmt.Errorf("%s: unable to register audit gated filter: %w", op, err) } @@ -280,12 +327,11 @@ func NewEventer(log hclog.Logger, c EventerConfig, opt ...Option) (*Eventer, err err = e.broker.RegisterPipeline(eventlogger.Pipeline{ EventType: eventlogger.EventType(p.eventType), PipelineID: eventlogger.PipelineID(pipeId), - NodeIDs: []eventlogger.NodeID{gatedFilterNodeId, p.fmtId, p.sinkId}, + NodeIDs: []eventlogger.NodeID{p.gateId, p.fmtId, p.sinkId}, }) 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 { @@ -320,36 +366,37 @@ func NewEventer(log hclog.Logger, c EventerConfig, opt ...Option) (*Eventer, err sysNodeIds = append(sysNodeIds, p.sinkId) } - // TODO(jimlambrt) go-eventlogger SetSuccessThreshold currently does not - // specify which sink passed and which hasn't so we are unable to - // support multiple sinks with different delivery guarantees - if c.AuditDelivery == Enforced { - 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) - } - } - if c.ObservationDelivery == Enforced { - 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) - } - } - if c.SysEventsDelivery == Enforced { - err = e.broker.SetSuccessThreshold(eventlogger.EventType(SystemType), len(sysNodeIds)) - if err != nil { - return nil, fmt.Errorf("%s: failed to set success threshold for system events: %w", op, err) - } - } // always enforce delivery of errors 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) } + e.auditPipelines = append(e.auditPipelines, auditPipelines...) + e.errPipelines = append(e.errPipelines, errPipelines...) + e.observationPipelines = append(e.observationPipelines, observationPipelines...) + return e, nil } +func DefaultEventerConfig() *EventerConfig { + return &EventerConfig{ + AuditEnabled: false, + ObservationsEnabled: true, + SysEventsEnabled: true, + Sinks: []SinkConfig{DefaultSink()}, + } +} + +func DefaultSink() SinkConfig { + return SinkConfig{ + Name: "default", + EventTypes: []Type{EveryType}, + Format: JSONSinkFormat, + SinkType: StderrSink, + } +} + // writeObservation writes/sends an Observation event. func (e *Eventer) writeObservation(ctx context.Context, event *observation) error { const op = "event.(Eventer).writeObservation" diff --git a/internal/observability/event/eventer_config.go b/internal/observability/event/eventer_config.go index 9fa3ea2e9..635510d7d 100644 --- a/internal/observability/event/eventer_config.go +++ b/internal/observability/event/eventer_config.go @@ -6,26 +6,16 @@ import ( // EventerConfig supplies all the configuration needed to create/config an Eventer. type EventerConfig struct { - AuditDelivery DeliveryGuarantee // AuditDelivery specifies the delivery guarantees for audit events (enforced or best effort). - ObservationDelivery DeliveryGuarantee // ObservationDelivery specifies the delivery guarantees for observation events (enforced or best effort). - SysEventsDelivery DeliveryGuarantee // SysEventsDelivery specifies the delivery guarantees for system events (enforced or best effort). - AuditEnabled bool // AuditEnabled specifies if audit events should be emitted. - ObservationsEnabled bool // ObservationsEnabled specifies if observation events should be emitted. - SysEventsEnabled bool // SysEventsEnabled specifies if sysevents should be emitted. - Sinks []SinkConfig // Sinks are all the configured sinks + AuditEnabled bool `hcl:"audit_enabled"` // AuditEnabled specifies if audit events should be emitted. + ObservationsEnabled bool `hcl:"observations_enabled"` // ObservationsEnabled specifies if observation events should be emitted. + SysEventsEnabled bool `hcl:"sysevents_enabled"` // SysEventsEnabled specifies if sysevents should be emitted. + Sinks []SinkConfig `hcl:"sinks"` // Sinks are all the configured sinks } -// validate will validate the config. BTW, a config isn't required to have any +// Validate will Validate the config. A config isn't required to have any // sinks to be valid. -func (c *EventerConfig) validate() error { - const op = "event.(EventerConfig).validate" - - if err := c.AuditDelivery.validate(); err != nil { - return fmt.Errorf("%s: %w", op, err) - } - if err := c.ObservationDelivery.validate(); err != nil { - return fmt.Errorf("%s: %w", op, err) - } +func (c *EventerConfig) Validate() error { + const op = "event.(EventerConfig).Validate" for i, s := range c.Sinks { if err := s.validate(); err != nil { return fmt.Errorf("%s: sink %d is invalid: %w", op, i, err) diff --git a/internal/observability/event/eventer_config_test.go b/internal/observability/event/eventer_config_test.go index 6b40eb775..1b63133cb 100644 --- a/internal/observability/event/eventer_config_test.go +++ b/internal/observability/event/eventer_config_test.go @@ -14,34 +14,6 @@ func TestEventerConfig_validate(t *testing.T) { wantErrIs error wantErrContains string }{ - { - name: "invalid-audit-delivery", - c: EventerConfig{ - AuditDelivery: "invalid", - }, - wantErrIs: ErrInvalidParameter, - wantErrContains: "is not a valid delivery guarantee", - }, - { - name: "valid-audit-delivery", - c: EventerConfig{ - AuditDelivery: Enforced, - }, - }, - { - name: "invalid-observation-delivery", - c: EventerConfig{ - ObservationDelivery: "invalid", - }, - wantErrIs: ErrInvalidParameter, - wantErrContains: "is not a valid delivery guarantee", - }, - { - name: "valid-observation-delivery", - c: EventerConfig{ - ObservationDelivery: Enforced, - }, - }, { name: "invalid-sink", c: EventerConfig{ @@ -63,7 +35,7 @@ func TestEventerConfig_validate(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { assert, require := assert.New(t), require.New(t) - err := tt.c.validate() + err := tt.c.Validate() if tt.wantErrIs != nil { require.Error(err) assert.ErrorIs(err, tt.wantErrIs) diff --git a/internal/observability/event/eventer_retry_test.go b/internal/observability/event/eventer_retry_test.go index 40ee3e695..01d5ee9e8 100644 --- a/internal/observability/event/eventer_retry_test.go +++ b/internal/observability/event/eventer_retry_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "os" + "sync" "testing" "github.com/hashicorp/eventlogger" @@ -19,7 +20,11 @@ func TestEventer_retrySend(t *testing.T) { testConfig := TestEventerConfig(t, "TestEventer_retrySend") - eventer, err := NewEventer(hclog.Default(), testConfig.EventerConfig) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + }) + eventer, err := NewEventer(testLogger, testLock, 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 233d11e6a..0ba6e8e32 100644 --- a/internal/observability/event/eventer_test.go +++ b/internal/observability/event/eventer_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "strings" + "sync" "testing" "github.com/hashicorp/eventlogger" @@ -16,55 +17,80 @@ func Test_InitSysEventer(t *testing.T) { // this test and its subtests cannot be run in parallel because of it's // dependency on the sysEventer testConfig := TestEventerConfig(t, "InitSysEventer") + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + }) + testEventer, err := NewEventer(testLogger, testLock, testConfig.EventerConfig) + require.NoError(t, err) tests := []struct { name string log hclog.Logger - config EventerConfig + lock *sync.Mutex + opt []Option want *Eventer wantErrIs error }{ - { - name: "missing-hclog", - config: testConfig.EventerConfig, + name: "missing-both-eventer-and-config", wantErrIs: ErrInvalidParameter, }, { - name: "success", - config: testConfig.EventerConfig, - log: hclog.Default(), + name: "missing-hclog", + opt: []Option{WithEventerConfig(&testConfig.EventerConfig)}, + lock: testLock, + wantErrIs: ErrInvalidParameter, + }, + { + name: "missing-lock", + opt: []Option{WithEventerConfig(&testConfig.EventerConfig)}, + log: testLogger, + wantErrIs: ErrInvalidParameter, + }, + { + name: "success-with-config", + opt: []Option{WithEventerConfig(&testConfig.EventerConfig)}, + log: testLogger, + lock: testLock, want: &Eventer{ - logger: hclog.Default(), + logger: testLogger, conf: testConfig.EventerConfig, }, }, { - name: "success-with-default-config", - config: EventerConfig{}, - log: hclog.Default(), + name: "success-with-default-config", + opt: []Option{WithEventerConfig(&EventerConfig{})}, + log: testLogger, + lock: testLock, want: &Eventer{ - logger: hclog.Default(), + logger: testLogger, conf: EventerConfig{ Sinks: []SinkConfig{ { Name: "default", EventTypes: []Type{EveryType}, Format: JSONSinkFormat, - SinkType: StdoutSink, + SinkType: StderrSink, }, }, }, }, }, + { + name: "success-with-eventer", + opt: []Option{WithEventer(testEventer)}, + log: testLogger, + lock: testLock, + want: testEventer, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { defer TestResetSystEventer(t) assert, require := assert.New(t), require.New(t) - - err := InitSysEventer(tt.log, tt.config) + err := InitSysEventer(tt.log, tt.lock, tt.opt...) got := SysEventer() if tt.wantErrIs != nil { require.Nil(got) @@ -76,6 +102,9 @@ func Test_InitSysEventer(t *testing.T) { require.NotNil(got) tt.want.broker = got.broker tt.want.flushableNodes = got.flushableNodes + tt.want.auditPipelines = got.auditPipelines + tt.want.errPipelines = got.errPipelines + tt.want.observationPipelines = got.observationPipelines assert.Equal(tt.want, got) }) } @@ -85,7 +114,11 @@ func TestEventer_writeObservation(t *testing.T) { t.Parallel() ctx := context.Background() testSetup := TestEventerConfig(t, "TestEventer_writeObservation") - eventer, err := NewEventer(hclog.Default(), testSetup.EventerConfig) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + }) + eventer, err := NewEventer(testLogger, testLock, testSetup.EventerConfig) require.NoError(t, err) testHeader := map[string]interface{}{"name": "header"} @@ -134,14 +167,16 @@ func TestEventer_writeObservation(t *testing.T) { t.Run("e2e", func(t *testing.T) { require := require.New(t) - logger := hclog.New(&hclog.LoggerOptions{ - Name: "test", - }) c := EventerConfig{ ObservationsEnabled: true, } - // with no defined config, it will default to a stdout sink - e, err := NewEventer(logger, c) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + // with no defined config, it will default to a stderr sink + e, err := NewEventer(testLogger, testLock, c) require.NoError(err) m := map[string]interface{}{ @@ -159,7 +194,12 @@ func TestEventer_writeAudit(t *testing.T) { t.Parallel() ctx := context.Background() testSetup := TestEventerConfig(t, "Test_NewEventer") - eventer, err := NewEventer(hclog.Default(), testSetup.EventerConfig) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + eventer, err := NewEventer(testLogger, testLock, testSetup.EventerConfig) require.NoError(t, err) testAudit, err := newAudit( @@ -214,7 +254,12 @@ func TestEventer_writeError(t *testing.T) { t.Parallel() ctx := context.Background() testSetup := TestEventerConfig(t, "Test_NewEventer") - eventer, er := NewEventer(hclog.Default(), testSetup.EventerConfig) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + eventer, er := NewEventer(testLogger, testLock, testSetup.EventerConfig) require.NoError(t, er) testError, er := newError("TestEventer_writeError", fmt.Errorf("%s: no msg: test", ErrIo)) @@ -264,13 +309,20 @@ func Test_NewEventer(t *testing.T) { t.Parallel() testSetup := TestEventerConfig(t, "Test_NewEventer") - testSetupWithOpts := TestEventerConfig(t, "Test_NewEventer", testWithAuditSink(), testWithObservationSink(), testWithSysSink()) + testSetupWithOpts := TestEventerConfig(t, "Test_NewEventer", TestWithAuditSink(t), TestWithObservationSink(t), testWithSysSink(t)) + + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) tests := []struct { name string config EventerConfig opts []Option logger hclog.Logger + lock *sync.Mutex want *Eventer wantRegistered []string wantPipelines []string @@ -278,45 +330,45 @@ func Test_NewEventer(t *testing.T) { wantErrIs error }{ { - name: "invalid-config", - config: EventerConfig{ - AuditDelivery: "invalid", - }, - logger: hclog.Default(), + name: "missing-logger", + config: testSetup.EventerConfig, + lock: testLock, wantErrIs: ErrInvalidParameter, }, { - name: "missing-logger", + name: "missing-lock", config: testSetup.EventerConfig, + logger: testLogger, wantErrIs: ErrInvalidParameter, }, { name: "success-with-default-config", config: EventerConfig{}, - logger: hclog.Default(), + logger: testLogger, + lock: testLock, want: &Eventer{ - logger: hclog.Default(), + logger: testLogger, conf: EventerConfig{ Sinks: []SinkConfig{ { Name: "default", EventTypes: []Type{EveryType}, Format: JSONSinkFormat, - SinkType: StdoutSink, + SinkType: StderrSink, }, }, }, }, wantRegistered: []string{ "json", // fmt for everything - "stdout", // stdout - "gated-observation", // stdout - "gated-audit", // stdout + "stderr", // stderr + "gated-observation", // stderr + "gated-audit", // stderr }, wantPipelines: []string{ - "audit", // stdout - "observation", // stdout - "error", // stdout + "audit", // stderr + "observation", // stderr + "error", // stderr "system", // stderr }, wantThresholds: map[eventlogger.EventType]int{ @@ -326,16 +378,17 @@ func Test_NewEventer(t *testing.T) { { name: "testSetup", config: testSetup.EventerConfig, - logger: hclog.Default(), + logger: testLogger, + lock: testLock, want: &Eventer{ - logger: hclog.Default(), + logger: testLogger, conf: testSetup.EventerConfig, }, wantRegistered: []string{ "json", // fmt for everything - "stdout", // stdout - "gated-observation", // stdout - "gated-audit", // stdout + "stderr", // stderr + "gated-observation", // stderr + "gated-audit", // stderr "tmp-all-events", // every-type-file-sync "gated-observation", // every-type-file-sync "gated-audit", // every-type-file-sync @@ -343,34 +396,33 @@ func Test_NewEventer(t *testing.T) { }, wantPipelines: []string{ "audit", // every-type-file-sync - "audit", // stdout + "audit", // stderr "observation", // every-type-file-sync - "observation", // stdout + "observation", // stderr "error", // every-type-file-sync - "error", // stdout + "error", // stderr "error", // error-file-sink "system", // stderr "system", // stderr }, wantThresholds: map[eventlogger.EventType]int{ - "audit": 2, - "error": 3, - "observation": 2, + "error": 3, }, }, { name: "testSetup-with-all-opts", config: testSetupWithOpts.EventerConfig, - logger: hclog.Default(), + logger: testLogger, + lock: testLock, want: &Eventer{ - logger: hclog.Default(), + logger: testLogger, conf: testSetupWithOpts.EventerConfig, }, wantRegistered: []string{ "json", // fmt for everything - "stdout", // stdout - "gated-observation", // stdout - "gated-audit", // stdout + "stderr", // stderr + "gated-observation", // stderr + "gated-audit", // stderr "tmp-all-events", // every-type-file-sync "gated-observation", // every-type-file-sync "gated-audit", // every-type-file-sync @@ -383,11 +435,11 @@ func Test_NewEventer(t *testing.T) { }, wantPipelines: []string{ "audit", // every-type-file-sync - "audit", // stdout + "audit", // stderr "observation", // every-type-file-sync - "observation", // stdout + "observation", // stderr "error", // every-type-file-sync - "error", // stdout + "error", // stderr "error", // error-file-sink "audit", // audit-file-sink "observation", // observation-file-sink @@ -396,9 +448,7 @@ func Test_NewEventer(t *testing.T) { "system", // sys-file-sink }, wantThresholds: map[eventlogger.EventType]int{ - "audit": 3, - "error": 3, - "observation": 3, + "error": 3, }, }, } @@ -407,7 +457,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.config, testWithBroker(testBroker)) + got, err := NewEventer(tt.logger, tt.lock, tt.config, TestWithBroker(t, testBroker)) if tt.wantErrIs != nil { require.Error(err) require.Nil(got) @@ -418,30 +468,69 @@ func Test_NewEventer(t *testing.T) { require.NotNil(got) tt.want.broker = got.broker tt.want.flushableNodes = got.flushableNodes + tt.want.auditPipelines = got.auditPipelines + tt.want.errPipelines = got.errPipelines + tt.want.observationPipelines = got.observationPipelines assert.Equal(tt.want, got) assert.Lenf(testBroker.registeredNodeIds, len(tt.wantRegistered), "got nodes: %q", testBroker.registeredNodeIds) + registeredNodeIds := map[string]bool{} + for _, id := range testBroker.registeredNodeIds { + registeredNodeIds[string(id)] = true + } for _, want := range tt.wantRegistered { found := false - for _, got := range testBroker.registeredNodeIds { - if strings.Contains(string(got), want) { + for got := range registeredNodeIds { + if strings.Contains(got, want) { found = true + delete(registeredNodeIds, got) break } } assert.Truef(found, "did not find %s in the registered nodes: %s", want, testBroker.registeredNodeIds) } + assert.Lenf(testBroker.pipelines, len(tt.wantPipelines), "got pipelines: %q", testBroker.pipelines) + registeredPipelines := map[string]eventlogger.Pipeline{} + gotAuditCnt := 0 + gotErrCnt := 0 + gotObservationCnt := 0 + for _, got := range testBroker.pipelines { + registeredPipelines[string(got.PipelineID)] = got + switch got.EventType { + case eventlogger.EventType(AuditType): + gotAuditCnt += 1 + case eventlogger.EventType(ErrorType): + gotErrCnt += 1 + case eventlogger.EventType(ObservationType): + gotObservationCnt += 1 + } + } + wantAuditCnt := 0 + wantErrCnt := 0 + wantObservationCnt := 0 for _, want := range tt.wantPipelines { + switch want { + case string(AuditType): + wantAuditCnt += 1 + case string(ErrorType): + wantErrCnt += 1 + case string(ObservationType): + wantObservationCnt += 1 + } found := false - for _, got := range testBroker.pipelines { + for id, got := range registeredPipelines { if strings.Contains(string(got.EventType), want) { found = true + delete(registeredPipelines, id) break } } assert.Truef(found, "did not find %s in the registered pipelines: %s", want, testBroker.pipelines) } + assert.Equal(wantAuditCnt, gotAuditCnt) + assert.Equal(wantErrCnt, gotErrCnt) + assert.Equal(wantObservationCnt, gotObservationCnt) assert.Equal(tt.wantThresholds, testBroker.successThresholds) }) @@ -453,7 +542,13 @@ func TestEventer_Reopen(t *testing.T) { t.Run("simple", func(t *testing.T) { assert, require := assert.New(t), require.New(t) - e, err := NewEventer(hclog.Default(), EventerConfig{}) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + + e, err := NewEventer(testLogger, testLock, EventerConfig{}) require.NoError(err) e.broker = nil @@ -469,8 +564,13 @@ func TestEventer_FlushNodes(t *testing.T) { t.Parallel() t.Run("simple", func(t *testing.T) { assert, require := assert.New(t), require.New(t) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) - e, err := NewEventer(hclog.Default(), EventerConfig{}) + e, err := NewEventer(testLogger, testLock, EventerConfig{}) require.NoError(err) node := &testFlushNode{} diff --git a/internal/observability/event/options.go b/internal/observability/event/options.go index 3d9534ee0..977579c90 100644 --- a/internal/observability/event/options.go +++ b/internal/observability/event/options.go @@ -8,7 +8,9 @@ import ( func getOpts(opt ...Option) options { opts := getDefaultOptions() for _, o := range opt { - o(&opts) + if o != nil { + o(&opts) + } } return opts } @@ -18,15 +20,17 @@ type Option func(*options) // options = how options are represented type options struct { - withId string - withDetails map[string]interface{} - withHeader map[string]interface{} - withFlush bool - withRequestInfo *RequestInfo - withNow time.Time - withRequest *Request - withResponse *Response - withAuth *Auth + withId string + withDetails map[string]interface{} + withHeader map[string]interface{} + withFlush bool + withRequestInfo *RequestInfo + withNow time.Time + withRequest *Request + withResponse *Response + withAuth *Auth + withEventer *Eventer + withEventerConfig *EventerConfig withBroker broker // test only option withAuditSink bool // test only option @@ -100,3 +104,17 @@ func WithAuth(a *Auth) Option { o.withAuth = a } } + +// WithEventer allows an optional eventer +func WithEventer(e *Eventer) Option { + return func(o *options) { + o.withEventer = e + } +} + +// WithEventer allows an optional eventer config +func WithEventerConfig(c *EventerConfig) Option { + return func(o *options) { + o.withEventerConfig = c + } +} diff --git a/internal/observability/event/options_test.go b/internal/observability/event/options_test.go index 17863ab8d..6fefd0960 100644 --- a/internal/observability/event/options_test.go +++ b/internal/observability/event/options_test.go @@ -84,4 +84,20 @@ func Test_GetOpts(t *testing.T) { testOpts.withAuth = auth assert.Equal(opts, testOpts) }) + t.Run("WithEventer", func(t *testing.T) { + assert := assert.New(t) + eventer := Eventer{} + opts := getOpts(WithEventer(&eventer)) + testOpts := getDefaultOptions() + testOpts.withEventer = &eventer + assert.Equal(opts, testOpts) + }) + t.Run("WithEventerConfig", func(t *testing.T) { + assert := assert.New(t) + c := EventerConfig{} + opts := getOpts(WithEventerConfig(&c)) + testOpts := getDefaultOptions() + testOpts.withEventerConfig = &c + assert.Equal(opts, testOpts) + }) } diff --git a/internal/observability/event/serialized_writer.go b/internal/observability/event/serialized_writer.go new file mode 100644 index 000000000..54a8e7b9d --- /dev/null +++ b/internal/observability/event/serialized_writer.go @@ -0,0 +1,39 @@ +package event + +import ( + "bytes" + "io" + "sync" + + "github.com/hashicorp/boundary/internal/errors" +) + +// serializedWriter uses a mutext to serializes all writes to its io.Writer +type serializedWriter struct { + l *sync.Mutex + w io.Writer +} + +// Write uses a mutex to serialize all writes +func (s *serializedWriter) Write(p []byte) (int, error) { + const op = "event.(serializedWriter).Write" + if s == nil { + return 0, errors.New(errors.InvalidParameter, op, "missing serialized writer") + } + if s.l == nil { + return 0, errors.New(errors.InvalidParameter, op, "missing lock") + } + if s.w == nil { + return 0, errors.New(errors.InvalidParameter, op, "missing writer") + } + + s.l.Lock() + defer s.l.Unlock() + reader := bytes.NewReader(p) + + n, err := reader.WriteTo(s.w) + if err != nil { + return 0, err + } + return int(n), err +} diff --git a/internal/observability/event/serialized_writer_test.go b/internal/observability/event/serialized_writer_test.go new file mode 100644 index 000000000..77de833ae --- /dev/null +++ b/internal/observability/event/serialized_writer_test.go @@ -0,0 +1,78 @@ +package event + +import ( + "os" + "sync" + "testing" + + "github.com/hashicorp/boundary/internal/errors" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type testBadWriter struct{} + +func (b *testBadWriter) Write(p []byte) (int, error) { + const op = "event.(testBadWriter).Write" + return 0, errors.New(errors.Internal, op, "write failed") +} + +func TestSerializedWriter_Write(t *testing.T) { + + tests := []struct { + name string + s *serializedWriter + wantErrMatch *errors.Template + wantErrContains string + }{ + { + name: "missing-serializedWriter", + wantErrMatch: errors.T(errors.InvalidParameter), + wantErrContains: "missing serialized writer", + }, + { + name: "missing-writer", + s: &serializedWriter{ + l: new(sync.Mutex), + }, + wantErrMatch: errors.T(errors.InvalidParameter), + wantErrContains: "missing writer", + }, + { + name: "missing-lock", + s: &serializedWriter{ + w: os.Stderr, + }, + wantErrMatch: errors.T(errors.InvalidParameter), + wantErrContains: "missing lock", + }, + { + name: "write-error", + s: &serializedWriter{ + w: &testBadWriter{}, + l: new(sync.Mutex), + }, + wantErrMatch: errors.T(errors.Internal), + wantErrContains: "write failed", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + + n, err := tt.s.Write([]byte("fido")) + if tt.wantErrMatch != nil { + require.Error(err) + require.Empty(n) + assert.Truef(errors.Match(tt.wantErrMatch, err), "got %q and wanted %q", tt.wantErrMatch.Code, err) + if tt.wantErrContains != "" { + assert.Contains(err.Error(), tt.wantErrContains) + } + return + } + require.NoError(err) + assert.NotEmpty(n) + }) + } + +} diff --git a/internal/observability/event/sink_config.go b/internal/observability/event/sink_config.go index 58f51b07d..12e680e0d 100644 --- a/internal/observability/event/sink_config.go +++ b/internal/observability/event/sink_config.go @@ -7,15 +7,16 @@ import ( // SinkConfig defines the configuration for a Eventer sink type SinkConfig struct { - Name string // Name defines a name for the sink. - EventTypes []Type // 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. - SinkType SinkType // SinkType defines the type of sink (StdoutSink or FileSink) - Format SinkFormat // Format defines the format for the sink (JSONSinkFormat) - Path string // Path defines the file path for the sink - FileName string // FileName defines the file name for the sink - RotateBytes int // RotateByes defines the number of bytes that should trigger rotation of a FileSink - RotateDuration time.Duration // RotateDuration defines how often a FileSink should be rotated - RotateMaxFiles int // RotateMaxFiles defines how may historical rotated files should be kept for a FileSink + 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. + 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) + 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 + RotateDuration time.Duration `hcl:"rotate_duration"` // RotateDuration defines how often a FileSink should be rotated + RotateMaxFiles int `hcl:"rotate_max_files"` // RotateMaxFiles defines how may historical rotated files should be kept for a FileSink } func (sc *SinkConfig) validate() error { @@ -23,7 +24,7 @@ func (sc *SinkConfig) validate() error { if err := sc.SinkType.validate(); err != nil { return fmt.Errorf("%s: %w", op, err) } - if err := sc.Format.validate(); err != nil { + if err := sc.Format.Validate(); err != nil { return fmt.Errorf("%s: %w", op, err) } if sc.SinkType == FileSink && sc.FileName == "" { diff --git a/internal/observability/event/sink_format.go b/internal/observability/event/sink_format.go index 6d4b406df..5712f8c66 100644 --- a/internal/observability/event/sink_format.go +++ b/internal/observability/event/sink_format.go @@ -10,8 +10,8 @@ const ( type SinkFormat string // SinkFormat defines the formatting for a sink in a config file stanza (json) -func (f SinkFormat) validate() error { - const op = "event.(SinkFormat).validate" +func (f SinkFormat) Validate() error { + const op = "event.(SinkFormat).Validate" switch f { case JSONSinkFormat: return nil diff --git a/internal/observability/event/sink_type.go b/internal/observability/event/sink_type.go index 889a22468..1b53e6ad4 100644 --- a/internal/observability/event/sink_type.go +++ b/internal/observability/event/sink_type.go @@ -5,16 +5,16 @@ import ( ) const ( - StdoutSink SinkType = "stdout" // StdoutSink is written to stdout + StderrSink SinkType = "stderr" // StderrSink is written to stderr FileSink SinkType = "file" // FileSink is written to a file ) -type SinkType string // SinkType defines the type of sink in a config stanza (file, stdout) +type SinkType string // SinkType defines the type of sink in a config stanza (file, stderr) func (t SinkType) validate() error { const op = "event.(SinkType).validate" switch t { - case StdoutSink, FileSink: + case StderrSink, FileSink: return nil default: return fmt.Errorf("%s: '%s' is not a valid sink type: %w", op, t, ErrInvalidParameter) diff --git a/internal/observability/event/testing.go b/internal/observability/event/testing.go index aacf5865c..ae1a7fd6c 100644 --- a/internal/observability/event/testing.go +++ b/internal/observability/event/testing.go @@ -4,10 +4,10 @@ import ( "context" "io/ioutil" "os" - "sync" "testing" "time" + "github.com/hashicorp/boundary/globals" "github.com/hashicorp/boundary/internal/gen/controller/api/resources/groups" pbs "github.com/hashicorp/boundary/internal/gen/controller/api/services" "github.com/hashicorp/eventlogger" @@ -15,21 +15,40 @@ import ( "google.golang.org/protobuf/types/known/wrapperspb" ) -var testSysEventerLock sync.Mutex +// TestEnableEventing will enable eventing via it's envvar +// globals.BOUNDARY_DEVELOPER_ENABLE_EVENTS +func TestEnableEventing(t *testing.T, enable bool) { + t.Helper() + if enable { + os.Setenv(globals.BOUNDARY_DEVELOPER_ENABLE_EVENTS, "true") + t.Cleanup(func() { + defer os.Unsetenv(globals.BOUNDARY_DEVELOPER_ENABLE_EVENTS) + }) + } else { + os.Unsetenv(globals.BOUNDARY_DEVELOPER_ENABLE_EVENTS) + } +} + +// TestGetEventerConfig is a test accessor for the eventer's config +func TestGetEventerConfig(t *testing.T, e *Eventer) EventerConfig { + t.Helper() + return e.conf +} // TestResetSysEventer will reset event.syseventer to an uninitialized state. func TestResetSystEventer(t *testing.T) { t.Helper() - testSysEventerLock.Lock() - defer testSysEventerLock.Unlock() - sysEventerOnce = sync.Once{} + sysEventerLock.Lock() + defer sysEventerLock.Unlock() sysEventer = nil } type TestConfig struct { - EventerConfig EventerConfig - AllEvents *os.File - ErrorEvents *os.File + EventerConfig EventerConfig + AllEvents *os.File + ErrorEvents *os.File + ObservationEvents *os.File + AuditEvents *os.File } // TestEventerConfig creates a test config and registers a cleanup func for its @@ -51,9 +70,7 @@ func TestEventerConfig(t *testing.T, testName string, opt ...Option) TestConfig c := TestConfig{ EventerConfig: EventerConfig{ ObservationsEnabled: true, - ObservationDelivery: Enforced, AuditEnabled: true, - AuditDelivery: Enforced, Sinks: []SinkConfig{ { Name: "every-type-file-sink", @@ -64,8 +81,8 @@ func TestEventerConfig(t *testing.T, testName string, opt ...Option) TestConfig FileName: tmpAllFile.Name(), }, { - Name: "stdout", - SinkType: StdoutSink, + Name: "stderr", + SinkType: StderrSink, EventTypes: []Type{EveryType}, Format: JSONSinkFormat, }, @@ -97,6 +114,7 @@ func TestEventerConfig(t *testing.T, testName string, opt ...Option) TestConfig Path: "./", FileName: tmpFile.Name(), }) + c.AuditEvents = tmpFile } if opts.withObservationSink { tmpFile, err := ioutil.TempFile("./", "tmp-observation-"+testName) @@ -112,6 +130,7 @@ func TestEventerConfig(t *testing.T, testName string, opt ...Option) TestConfig Path: "./", FileName: tmpFile.Name(), }) + c.ObservationEvents = tmpFile } if opts.withSysSink { tmpFile, err := ioutil.TempFile("./", "tmp-sysevents-"+testName) @@ -177,29 +196,33 @@ func testResponse(t *testing.T) *Response { } } -// testWithBroker is an unexported and a test option for passing in an optional broker -func testWithBroker(b broker) Option { +// TestWithBroker is an unexported and a test option for passing in an optional broker +func TestWithBroker(t *testing.T, b broker) Option { + t.Helper() return func(o *options) { o.withBroker = b } } -// testWithObservationSink is an unexported and a test option -func testWithObservationSink() Option { +// TestWithObservationSink is an unexported and a test option +func TestWithObservationSink(t *testing.T) Option { + t.Helper() return func(o *options) { o.withObservationSink = true } } -// testWithAuditSink is an unexported and a test option -func testWithAuditSink() Option { +// TestWithAuditSink is an unexported and a test option +func TestWithAuditSink(t *testing.T) Option { + t.Helper() return func(o *options) { o.withAuditSink = true } } // testWithSysSink is an unexported and a test option -func testWithSysSink() Option { +func testWithSysSink(t *testing.T) Option { + t.Helper() return func(o *options) { o.withSysSink = true } diff --git a/internal/servers/common/handler.go b/internal/servers/common/handler.go new file mode 100644 index 000000000..66a937b9f --- /dev/null +++ b/internal/servers/common/handler.go @@ -0,0 +1,220 @@ +package common + +import ( + "context" + "fmt" + "net/http" + "time" + + "github.com/hashicorp/boundary/internal/auth" + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/kms" + "github.com/hashicorp/boundary/internal/observability/event" + "github.com/hashicorp/go-hclog" + "github.com/hashicorp/vault/sdk/helper/base62" +) + +// generatedTraceId returns a boundary generated TraceId or "" if an error occurs when generating +// the id. +func generatedTraceId(ctx context.Context) string { + t, err := base62.Random(20) + if err != nil { + return "" + } + return fmt.Sprintf("gtraceid_%s", t) +} + +type writerWrapper struct { + http.ResponseWriter + statusCode int +} + +func (w *writerWrapper) WriteHeader(code int) { + w.statusCode = code + w.ResponseWriter.WriteHeader(code) +} + +func (w *writerWrapper) StatusCode() int { + return w.statusCode +} + +// WrapWithOptionals will test the wrap ResponseWriter and test it for various +// optional interfaces: http.Flusher, http.Pusher and http.Hijacker. It return +// an http.ResponseWriter that satisfies those optional interfaces. +// +// See: https://medium.com/@cep21/interface-wrapping-method-erasure-c523b3549912 +func WrapWithOptionals(with *writerWrapper, wrap http.ResponseWriter) (http.ResponseWriter, error) { + const op = "common.WrapWithOptionals" + if with == nil { + return nil, errors.New(errors.InvalidParameter, op, "missing writer wrapper") + } + if wrap == nil { + return nil, errors.New(errors.InvalidParameter, op, "missing response writer") + } + flusher, _ := wrap.(http.Flusher) + pusher, _ := wrap.(http.Pusher) + hijacker, _ := wrap.(http.Hijacker) + switch { + case flusher == nil && pusher == nil && hijacker == nil: + return with, nil + case flusher != nil && pusher == nil && hijacker == nil: + return struct { + *writerWrapper + http.Flusher + }{with, flusher}, nil + case flusher == nil && pusher != nil && hijacker == nil: + return struct { + *writerWrapper + http.Pusher + }{with, pusher}, nil + case flusher == nil && pusher == nil && hijacker != nil: + return struct { + *writerWrapper + http.Hijacker + }{with, hijacker}, nil + case flusher != nil && pusher != nil && hijacker == nil: + return struct { + *writerWrapper + http.Flusher + http.Pusher + }{with, flusher, pusher}, nil + case flusher != nil && pusher == nil && hijacker != nil: + return struct { + *writerWrapper + http.Flusher + http.Hijacker + }{with, flusher, hijacker}, nil + case flusher == nil && pusher != nil && hijacker != nil: + return struct { + *writerWrapper + http.Pusher + http.Hijacker + }{with, pusher, hijacker}, nil + default: + return struct { + *writerWrapper + http.Flusher + http.Pusher + http.Hijacker + }{with, flusher, pusher, hijacker}, nil + } +} + +// WrapWithEventsHandler will wrap the provided http.Handler with a +// handler that adds an Eventer to the request context and starts/flushes gated +// events of type: observation and audit +func WrapWithEventsHandler(h http.Handler, e *event.Eventer, logger hclog.Logger, kms *kms.Kms) (http.Handler, error) { + const op = "common.WrapWithEventsHandler" + if h == nil { + return nil, errors.New(errors.InvalidParameter, op, "missing handler") + } + if e == nil { + return nil, errors.New(errors.InvalidParameter, op, "missing eventer") + } + if logger == nil { + return nil, errors.New(errors.InvalidParameter, op, "missing logger") + } + if kms == nil { + return nil, errors.New(errors.InvalidParameter, op, "missing kms") + } + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + publicId, _, _ := auth.GetTokenFromRequest(logger, kms, r) + + var err error + info := &event.RequestInfo{ + Id: generatedTraceId(ctx), + PublicId: publicId, + Method: r.Method, + Path: r.URL.RequestURI(), + } + ctx, err = event.NewRequestInfoContext(ctx, info) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + logger.Trace("unable to create context with request info", "method", r.Method, "url", r.URL.RequestURI(), "error", err) + return + } + ctx, err = event.NewEventerContext(ctx, e) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + logger.Trace("unable to create context with eventer", "method", r.Method, "url", r.URL.RequestURI(), "error", err) + return + } + + // Set the context back on the request + r = r.WithContext(ctx) + + { + method := r.Method + url := r.URL.RequestURI() + start := time.Now() + if err := startGatedEvents(ctx, method, url, start); err != nil { + w.WriteHeader(http.StatusInternalServerError) + logger.Trace("unable to start gated events", "method", r.Method, "url", r.URL.RequestURI(), "error", err) + return + } + + wrapper, err := WrapWithOptionals(&writerWrapper{w, http.StatusOK}, w) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + logger.Trace("unable to wrap handler with optional interfaces", "method", r.Method, "url", r.URL.RequestURI(), "error", err) + return + } + h.ServeHTTP(wrapper, r) + + i, _ := wrapper.(interface{ StatusCode() int }) + if err := flushGatedEvents(ctx, method, url, i.StatusCode(), start); err != nil { + w.WriteHeader(http.StatusInternalServerError) + logger.Trace("unable to flush gated events", "method", r.Method, "url", r.URL.RequestURI(), "error", err) + return + } + } + }), nil +} + +// startGatedEvents will send "initial" events for all event types which are +// gated. This approach is okay since if at event type isn't enabled sending +// an event for it becomes a no op in the eventlogger.Broker +func startGatedEvents(ctx context.Context, method, url string, startTime time.Time) error { + const op = "common.startGatedEvents" + if startTime.IsZero() { + startTime = time.Now() + } + err := event.WriteObservation(ctx, "handler", event.WithHeader(map[string]interface{}{ + "start": startTime, + })) + if err != nil { + return errors.Wrap(err, op, errors.WithMsg("unable to write observation event")) + } + + err = event.WriteAudit(ctx, "handler") + if err != nil { + return errors.Wrap(err, op, errors.WithMsg("unable to write audit event")) + } + return nil +} + +// flushGatedEvents will send flush events for all event types which are +// gated. This approach is okay since if at event type isn't enabled sending +// an event for it becomes a no op in the eventlogger.Broker +func flushGatedEvents(ctx context.Context, method, url string, statusCode int, startTime time.Time) error { + const op = "common.flushGatedEvents" + stopTime := time.Now() + var latency float64 + if !startTime.IsZero() { + latency = float64(stopTime.Sub(startTime)) / float64(time.Millisecond) + } + err := event.WriteObservation(ctx, "handler", event.WithFlush(), event.WithHeader(map[string]interface{}{ + "stop": stopTime, + "latency-ms": latency, + "status": statusCode, + })) + if err != nil { + return errors.Wrap(err, op, errors.WithMsg("unable to write and flush observation event")) + } + err = event.WriteAudit(ctx, "handler", event.WithFlush()) + if err != nil { + return errors.Wrap(err, op, errors.WithMsg("unable to write and flush audit event")) + } + return nil +} diff --git a/internal/servers/common/handler_test.go b/internal/servers/common/handler_test.go new file mode 100644 index 000000000..f002508ad --- /dev/null +++ b/internal/servers/common/handler_test.go @@ -0,0 +1,586 @@ +package common + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io/ioutil" + "net" + "net/http" + "net/http/httptest" + "os" + "sync" + "testing" + "time" + + "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/kms" + "github.com/hashicorp/boundary/internal/observability/event" + "github.com/hashicorp/eventlogger" + "github.com/hashicorp/go-hclog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func Test_WrapWithOptionals(t *testing.T) { + t.Parallel() + w := httptest.NewRecorder() + testWriterWrapper := writerWrapper{w, 0} + + type testNoOptional struct { + http.ResponseWriter + } + type testPusherHijacker struct { + http.ResponseWriter + testHijacker + testPusher + } + type testPusherFlusher struct { + http.ResponseWriter + testPusher + testFlusher + } + type testFlusherHijacker struct { + http.ResponseWriter + testFlusher + testHijacker + } + + type testAll struct { + http.ResponseWriter + testFlusher + testHijacker + testPusher + } + + tests := []struct { + name string + with *writerWrapper + wrap http.ResponseWriter + wantFlusher bool + wantPusher bool + wantHijacker bool + wantErrMatch *errors.Template + wantErrContains string + }{ + { + name: "missing-test-writer", + wrap: &testFlusher{}, + wantErrMatch: errors.T(errors.InvalidParameter), + wantErrContains: "missing writer wrapper", + }, + { + name: "missing-wrapper", + with: &testWriterWrapper, + wantErrMatch: errors.T(errors.InvalidParameter), + wantErrContains: "missing response writer", + }, + { + name: "success-no-optional", + with: &testWriterWrapper, + wrap: &testNoOptional{}, + }, + { + name: "success-flusher", + with: &testWriterWrapper, + wrap: &testFlusher{}, + wantFlusher: true, + }, + { + name: "success-flusher-hijacker", + with: &testWriterWrapper, + wrap: &testFlusherHijacker{}, + wantFlusher: true, + }, + { + name: "success-pusher", + with: &testWriterWrapper, + wrap: &testPusher{}, + wantPusher: true, + }, + { + name: "success-pusher-hijacker", + with: &testWriterWrapper, + wrap: &testPusherHijacker{}, + wantHijacker: true, + wantPusher: true, + }, + { + name: "success-pusher-flusher", + with: &testWriterWrapper, + wrap: &testPusherFlusher{}, + wantFlusher: true, + wantPusher: true, + }, + { + name: "success-hijacker", + with: &testWriterWrapper, + wrap: &testHijacker{}, + wantHijacker: true, + }, + { + name: "success-all", + with: &testWriterWrapper, + wrap: &testAll{}, + wantHijacker: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + wrapped, err := WrapWithOptionals(tt.with, tt.wrap) + if tt.wantErrMatch != nil { + require.Error(err) + assert.Nil(wrapped) + assert.Truef(errors.Match(tt.wantErrMatch, err), "wanted %q and got %q", tt.wantErrMatch.Code, err.Error()) + if tt.wantErrContains != "" { + assert.Contains(err.Error(), tt.wantErrContains) + } + return + } + require.NoError(err) + require.NotNil(wrapped) + _, ok := wrapped.(interface{ StatusCode() int }) + assert.Truef(ok, "wanted an response writer that satisfied the StatusCode interface") + if tt.wantPusher { + _, ok := wrapped.(http.Pusher) + assert.Truef(ok, "wanted an response writer that satisfied the http.Pusher interface") + } + if tt.wantHijacker { + _, ok := wrapped.(http.Hijacker) + assert.Truef(ok, "wanted an response writer that satisfied the http.Hijacker interface") + } + if tt.wantFlusher { + _, ok := wrapped.(http.Flusher) + assert.Truef(ok, "wanted an response writer that satisfied the http.Flusher interface") + } + }) + } +} + +func Test_WrapWithEventsHandler(t *testing.T) { + // this cannot run in parallel because it relies on envvar + // globals.BOUNDARY_DEVELOPER_ENABLE_EVENTS + event.TestEnableEventing(t, true) + wrapper := db.TestWrapper(t) + conn, _ := db.TestSetup(t, "postgres") + testKms := kms.TestKms(t, conn, wrapper) + + testHander := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusTeapot) + fmt.Fprintln(w, "I'm a little teapot short and stout") + }) + + c := event.TestEventerConfig(t, "Test_WrapWithEventsHandler", event.TestWithAuditSink(t), event.TestWithObservationSink(t)) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + testEventer, err := event.NewEventer(testLogger, testLock, c.EventerConfig) + require.NoError(t, err) + + tests := []struct { + name string + h http.Handler + e *event.Eventer + logger hclog.Logger + kms *kms.Kms + statusCode int + noEventJson bool + wantErrMatch *errors.Template + wantErrContains string + }{ + { + name: "missing handler", + e: testEventer, + logger: hclog.Default(), + kms: testKms, + wantErrMatch: errors.T(errors.InvalidParameter), + wantErrContains: "missing handler", + }, + { + name: "missing eventer", + h: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprintln(w, "Hello, client") + }), + logger: hclog.Default(), + kms: testKms, + wantErrMatch: errors.T(errors.InvalidParameter), + wantErrContains: "missing eventer", + }, + { + name: "missing logger", + h: testHander, + e: testEventer, + kms: testKms, + wantErrMatch: errors.T(errors.InvalidParameter), + wantErrContains: "missing logger", + }, + { + name: "missing kms", + h: testHander, + e: testEventer, + logger: hclog.Default(), + wantErrMatch: errors.T(errors.InvalidParameter), + wantErrContains: "missing kms", + }, + { + name: "audit-startGatedEvents", + h: testHander, + e: func() *event.Eventer { + b := &testMockBroker{errorOnSendAudit: true} + c := event.EventerConfig{AuditEnabled: true} + e, err := event.NewEventer(testLogger, testLock, c, event.TestWithBroker(t, b)) + require.NoError(t, err) + return e + }(), + logger: hclog.Default(), + kms: testKms, + statusCode: http.StatusInternalServerError, + noEventJson: true, + }, + { + name: "audit-flushGatedEvents", + h: testHander, + e: func() *event.Eventer { + b := &testMockBroker{errorOnFlush: true} + c := event.EventerConfig{AuditEnabled: true} + e, err := event.NewEventer(testLogger, testLock, c, event.TestWithBroker(t, b)) + require.NoError(t, err) + return e + }(), + logger: hclog.Default(), + kms: testKms, + statusCode: http.StatusTeapot, // this isn't ideal, but the write by the test handler will send an teapot status + noEventJson: true, + }, + { + name: "success", + h: testHander, + e: testEventer, + logger: hclog.Default(), + kms: testKms, + statusCode: http.StatusTeapot, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + got, err := WrapWithEventsHandler(tt.h, tt.e, tt.logger, tt.kms) + if tt.wantErrMatch != nil { + require.Error(err) + assert.Nil(got) + assert.Truef(errors.Match(tt.wantErrMatch, err), "wanted %q and got %q", tt.wantErrMatch.Code, err.Error()) + if tt.wantErrContains != "" { + assert.Contains(err.Error(), tt.wantErrContains) + } + return + } + require.NoError(err) + assert.NotNil(got) + + req, err := http.NewRequest("GET", "/greeting", nil) + require.NoError(err) + rr := httptest.NewRecorder() + got.ServeHTTP(rr, req) + assert.Equal(tt.statusCode, rr.Code) + + { // test that the got observation is what we wanted. + require.NotNil(c.ObservationEvents) + defer func() { _ = os.WriteFile(c.ObservationEvents.Name(), nil, 0o666) }() + b, err := ioutil.ReadFile(c.ObservationEvents.Name()) + assert.NoError(err) + + if tt.noEventJson { + assert.Lenf(b, 0, "expected no json for internal errors but got %s", string(b)) + return + } + got := &eventJson{} + err = json.Unmarshal(b, got) + require.NoErrorf(err, "json: %s", string(b)) + + actualJson, err := json.Marshal(got) + require.NoError(err) + + // set the got values to the wanted values that are either + // static or calculated in real-time + info := event.RequestInfo{ + Method: "GET", + Path: "/greeting", + Id: got.Payload["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), + } + wantJson := testJson(t, event.ObservationType, &info, event.Op(tt.name), got, hdr, nil) + assert.JSONEq(string(wantJson), string(actualJson)) + } + + { // test that the got audit is what we wanted. + require.NotNil(c.AuditEvents) + defer func() { _ = os.WriteFile(c.AuditEvents.Name(), nil, 0o666) }() + b, err := ioutil.ReadFile(c.AuditEvents.Name()) + assert.NoError(err) + + got := &eventJson{} + err = json.Unmarshal(b, got) + require.NoErrorf(err, "json: %s", string(b)) + + actualJson, err := json.Marshal(got) + require.NoError(err) + + // set the got values to the wanted values that are either + // static or calculated in real-time + info := event.RequestInfo{ + Method: "GET", + Path: "/greeting", + Id: got.Payload["id"].(string), + } + hdr := map[string]interface{}{ + "id": got.Payload["id"].(string), + "timestamp": got.Payload["timestamp"].(string), + "serialized_hmac": got.Payload["serialized_hmac"].(string), + } + wantJson := testJson(t, event.AuditType, &info, event.Op(tt.name), got, hdr, nil) + assert.JSONEq(string(wantJson), string(actualJson)) + + } + }) + } + +} + +func Test_startGatedEvents(t *testing.T) { + // this cannot run in parallel because it relies on envvar + // globals.BOUNDARY_DEVELOPER_ENABLE_EVENTS + event.TestEnableEventing(t, true) + testStartTime := time.Now() + tests := []struct { + name string + errOnAudit bool + errOnObservation bool + startTime time.Time + wantErrMatch *errors.Template + wantErrContains string + }{ + { + name: "audit-failed", + errOnAudit: true, + startTime: testStartTime, + wantErrMatch: errors.T(errors.Internal), + }, + { + name: "observation-failed", + errOnObservation: true, + wantErrMatch: errors.T(errors.Internal), + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + b := &testMockBroker{ + errorOnSendAudit: tt.errOnAudit, + errorOnSendObservation: tt.errOnObservation, + } + config := event.EventerConfig{ + AuditEnabled: true, + ObservationsEnabled: true, + } + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + e, err := event.NewEventer(testLogger, testLock, config, event.TestWithBroker(t, b)) + require.NoError(err) + ctx, err := event.NewEventerContext(context.Background(), e) + require.NoError(err) + err = startGatedEvents(ctx, "GET", "/hello", tt.startTime) + if tt.wantErrMatch != nil { + require.Error(err) + assert.Truef(errors.Match(tt.wantErrMatch, err), "wanted %q and got %q", tt.wantErrMatch.Code, err.Error()) + if tt.wantErrContains != "" { + assert.Contains(err.Error(), tt.wantErrContains) + } + return + } + require.NoError(err) + }) + } +} + +func Test_flushGatedEvents(t *testing.T) { + // this cannot run in parallel because it relies on envvar + // globals.BOUNDARY_DEVELOPER_ENABLE_EVENTS + event.TestEnableEventing(t, true) + testStartTime := time.Now() + tests := []struct { + name string + errOnAudit bool + errOnObservation bool + startTime time.Time + wantErrMatch *errors.Template + wantErrContains string + }{ + { + name: "audit-failed", + errOnAudit: true, + startTime: testStartTime, + wantErrMatch: errors.T(errors.Internal), + }, + { + name: "observation-failed", + errOnObservation: true, + wantErrMatch: errors.T(errors.Internal), + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + b := &testMockBroker{ + errorOnSendAudit: tt.errOnAudit, + errorOnSendObservation: tt.errOnObservation, + } + config := event.EventerConfig{ + AuditEnabled: true, + ObservationsEnabled: true, + } + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + e, err := event.NewEventer(testLogger, testLock, config, event.TestWithBroker(t, b)) + require.NoError(err) + ctx, err := event.NewEventerContext(context.Background(), e) + require.NoError(err) + err = flushGatedEvents(ctx, "GET", "/hello", 200, tt.startTime) + if tt.wantErrMatch != nil { + require.Error(err) + assert.Truef(errors.Match(tt.wantErrMatch, err), "wanted %q and got %q", tt.wantErrMatch.Code, err.Error()) + if tt.wantErrContains != "" { + assert.Contains(err.Error(), tt.wantErrContains) + } + return + } + require.NoError(err) + }) + } +} + +type testMockBroker struct { + errorOnSendAudit bool + errorOnSendObservation bool + errorOnFlush bool +} + +func (b *testMockBroker) Send(ctx context.Context, t eventlogger.EventType, payload interface{}) (eventlogger.Status, error) { + const op = "common.(testMockBroker).Send" + switch { + case b.errorOnFlush && payload.(eventlogger.Gateable).FlushEvent(): + return eventlogger.Status{}, errors.New(errors.Internal, op, "unable to flush event") + case b.errorOnSendAudit && t == eventlogger.EventType(event.AuditType): + return eventlogger.Status{}, errors.New(errors.Internal, op, "unable to send audit event") + case b.errorOnSendObservation && t == eventlogger.EventType(event.ObservationType): + return eventlogger.Status{}, errors.New(errors.Internal, op, "unable to send observation event") + } + return eventlogger.Status{}, nil +} +func (b *testMockBroker) Reopen(ctx context.Context) error { return nil } +func (b *testMockBroker) RegisterPipeline(def eventlogger.Pipeline) error { return nil } +func (b *testMockBroker) StopTimeAt(t time.Time) {} +func (b *testMockBroker) RegisterNode(id eventlogger.NodeID, node eventlogger.Node) error { return nil } +func (b *testMockBroker) SetSuccessThreshold(t eventlogger.EventType, successThreshold int) error { + return nil +} + +type eventJson struct { + CreatedAt string `json:"created_at"` + EventType string `json:"event_type"` + 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 { + t.Helper() + const ( + testAuditVersion = "v0.1" + testErrorVersion = "v0.1" + testObservationVersion = "v0.1" + ) + + require := require.New(t) + + var payload map[string]interface{} + 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, + }, + } + h := payload[event.HeaderField].(map[string]interface{}) + for k, v := range hdr { + h[k] = v + } + case event.AuditType: + payload = map[string]interface{}{ + event.IdField: got.Payload[event.IdField].(string), + event.RequestInfoField: reqInfo, + event.VersionField: testAuditVersion, + event.TypeField: event.ApiRequest, + } + for k, v := range hdr { + payload[k] = v + } + } + j := eventJson{ + CreatedAt: got.CreatedAt, + EventType: string(eventType), + Payload: payload, + } + + if details != nil { + details[event.OpField] = string(caller) + d := got.Payload[event.DetailsField].([]interface{})[0].(map[string]interface{}) + j.Payload[event.DetailsField] = []struct { + CreatedAt string `json:"created_at"` + Type string `json:"type"` + Payload map[string]interface{} `json:"payload"` + }{ + { + CreatedAt: d[event.CreatedAtField].(string), + Type: d[event.TypeField].(string), + Payload: details, + }, + } + } + b, err := json.Marshal(j) + require.NoError(err) + return b +} + +type testFlusher struct { + http.ResponseWriter +} + +func (t *testFlusher) Flush() {} + +type testPusher struct { + http.ResponseWriter +} + +func (t *testPusher) Push(target string, opts *http.PushOptions) error { return nil } + +type testHijacker struct { + http.ResponseWriter +} + +func (t *testHijacker) Hijack() (net.Conn, *bufio.ReadWriter, error) { return nil, nil, nil } diff --git a/internal/servers/controller/controller.go b/internal/servers/controller/controller.go index 4541b6108..c21f21e44 100644 --- a/internal/servers/controller/controller.go +++ b/internal/servers/controller/controller.go @@ -195,6 +195,11 @@ func (c *Controller) Shutdown(serversOnly bool) error { return fmt.Errorf("error stopping controller listeners: %w", err) } c.started.Store(false) + if c.conf.Eventer != nil { + if err := c.conf.Eventer.FlushNodes(context.Background()); err != nil { + return fmt.Errorf("error flushing controller eventer nodes: %w", err) + } + } return nil } diff --git a/internal/servers/controller/handler.go b/internal/servers/controller/handler.go index fa91c0555..e23784c99 100644 --- a/internal/servers/controller/handler.go +++ b/internal/servers/controller/handler.go @@ -18,6 +18,7 @@ import ( "github.com/hashicorp/boundary/internal/auth/oidc" "github.com/hashicorp/boundary/internal/gen/controller/api/services" "github.com/hashicorp/boundary/internal/requests" + "github.com/hashicorp/boundary/internal/servers/common" "github.com/hashicorp/boundary/internal/servers/controller/handlers/accounts" "github.com/hashicorp/boundary/internal/servers/controller/handlers/authmethods" "github.com/hashicorp/boundary/internal/servers/controller/handlers/credentiallibraries" @@ -29,7 +30,6 @@ import ( "github.com/hashicorp/boundary/sdk/strutil" "github.com/hashicorp/go-cleanhttp" "github.com/hashicorp/shared-secure-libs/configutil" - "github.com/hashicorp/vault/sdk/helper/base62" "google.golang.org/grpc/codes" "github.com/hashicorp/boundary/internal/servers/controller/handlers" @@ -64,8 +64,12 @@ func (c *Controller) handler(props HandlerProperties) (http.Handler, error) { commonWrappedHandler := wrapHandlerWithCommonFuncs(corsWrappedHandler, c, props) callbackInterceptingHandler := wrapHandlerWithCallbackInterceptor(commonWrappedHandler, c) printablePathCheckHandler := cleanhttp.PrintablePathCheckHandler(callbackInterceptingHandler, nil) + eventsHandler, err := common.WrapWithEventsHandler(printablePathCheckHandler, c.conf.Eventer, c.logger, c.kms) + if err != nil { + return nil, err + } - return printablePathCheckHandler, nil + return eventsHandler, nil } func handleGrpcGateway(c *Controller, props HandlerProperties) (http.Handler, error) { @@ -195,16 +199,6 @@ func handleGrpcGateway(c *Controller, props HandlerProperties) (http.Handler, er return mux, nil } -// generatedTraceId returns a boundary generated TraceId or "" if an error occurs when generating -// the id. -func generatedTraceId() string { - t, err := base62.Random(20) - if err != nil { - return "" - } - return fmt.Sprintf("gtraceid_%s", t) -} - func wrapHandlerWithCommonFuncs(h http.Handler, c *Controller, props HandlerProperties) http.Handler { var maxRequestDuration time.Duration var maxRequestSize int64 diff --git a/internal/servers/controller/testing.go b/internal/servers/controller/testing.go index 53df0564f..cd191f206 100644 --- a/internal/servers/controller/testing.go +++ b/internal/servers/controller/testing.go @@ -448,9 +448,14 @@ func NewTestController(t *testing.T, opts *TestControllerOpts) *TestController { if tc.b.Logger == nil { tc.b.Logger = hclog.New(&hclog.LoggerOptions{ Level: hclog.Trace, + Mutex: tc.b.StderrLock, }) } + if err := tc.b.SetupEventing(tc.b.Logger, tc.b.StderrLock, base.WithEventerConfig(opts.Config.Eventing)); err != nil { + t.Fatal(err) + } + if opts.Config.Controller == nil { opts.Config.Controller = new(config.Controller) } diff --git a/internal/servers/worker/worker.go b/internal/servers/worker/worker.go index 315215195..b72100ab3 100644 --- a/internal/servers/worker/worker.go +++ b/internal/servers/worker/worker.go @@ -140,6 +140,11 @@ func (w *Worker) Shutdown(skipListeners bool) error { } } w.started.Store(false) + if w.conf.Eventer != nil { + if err := w.conf.Eventer.FlushNodes(context.Background()); err != nil { + return fmt.Errorf("error flushing worker eventer nodes: %w", err) + } + } return nil }