CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Events: add node for formatting/filtering cloudevents (#1390)

Browse Source
This commit is contained in:
Jim authored and GitHub committed 2021-07-25 17:40:54 -04:00
1 parent 1ebb8efab6
commit 11124794bf
43 files changed
+1591 -320
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

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