integrate events into the servers (#1355)

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"
+)

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
+}

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)
+		})
+	}
+
+}

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 {

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)
+	})
+}

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)
+		})
+	}
+}

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

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

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.

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 {

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 {

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)

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
 }
 

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)
+		})
+
+	}
+}

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)

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(),

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
 }

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),

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"

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)

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)

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)

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{}

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
+	}
+}

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)
+	})
 }

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
+}

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)
+		})
+	}
+
+}

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 == "" {

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

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)

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
 	}

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
+}

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 }

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
 }
 

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

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)
 	}

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
 }