From 167161448be5c708bc1010bc5b967994388a06e1 Mon Sep 17 00:00:00 2001 From: Sepehr Date: Tue, 19 Sep 2023 17:16:37 -0700 Subject: [PATCH] Add write observation interceptors, add validation check, fix default deny filter --- internal/daemon/controller/gateway.go | 4 +- internal/daemon/controller/interceptor.go | 14 +++-- internal/daemon/controller/listeners.go | 4 +- internal/event/cloudevents_formatter_node.go | 13 ++--- .../event/cloudevents_formatter_node_test.go | 10 ++-- internal/event/context.go | 13 ++++- internal/event/context_test.go | 52 +++++++++++++++---- internal/event/event_observation.go | 3 ++ 8 files changed, 84 insertions(+), 29 deletions(-) diff --git a/internal/daemon/controller/gateway.go b/internal/daemon/controller/gateway.go index 931b625bb..a832184b0 100644 --- a/internal/daemon/controller/gateway.go +++ b/internal/daemon/controller/gateway.go @@ -116,9 +116,9 @@ func newGrpcServer( unaryCtxInterceptor, // populated requestInfo from headers into the request ctx errorInterceptor(ctx), // convert domain and api errors into headers for the http proxy subtypes.AttributeTransformerInterceptor(ctx), // convert to/from generic attributes from/to subtype specific attributes - auditRequestInterceptor(ctx), // before we get started, audit the request + eventsRequestInterceptor(ctx), // before we get started, send the required events with the request statusCodeInterceptor(ctx), // convert grpc codes into http status codes for the http proxy (can modify the resp) - auditResponseInterceptor(ctx), // as we finish, audit the response + eventsResponseInterceptor(ctx), // as we finish, send the required events with the response grpc_recovery.UnaryServerInterceptor( // recover from panics with a grpc internal error grpc_recovery.WithRecoveryHandlerContext(recoveryHandler()), ), diff --git a/internal/daemon/controller/interceptor.go b/internal/daemon/controller/interceptor.go index 7d749bd32..ca8c0cb8d 100644 --- a/internal/daemon/controller/interceptor.go +++ b/internal/daemon/controller/interceptor.go @@ -364,10 +364,10 @@ func isNil(i any) bool { return false } -func auditRequestInterceptor( +func eventsRequestInterceptor( _ context.Context, ) grpc.UnaryServerInterceptor { - const op = "controller.auditRequestInterceptor" + const op = "controller.eventsRequestInterceptor" return func(interceptorCtx context.Context, req any, _ *grpc.UnaryServerInfo, @@ -380,16 +380,19 @@ func auditRequestInterceptor( if err := event.WriteAudit(interceptorCtx, op, event.WithRequest(&event.Request{Details: clonedMsg})); err != nil { return req, status.Errorf(codes.Internal, "unable to write request msg audit: %s", err) } + if err := event.WriteObservation(interceptorCtx, op, event.WithRequest(&event.Request{Details: clonedMsg})); err != nil { + return req, status.Errorf(codes.Internal, "unable to write request msg observation: %s", err) + } } return handler(interceptorCtx, req) } } -func auditResponseInterceptor( +func eventsResponseInterceptor( _ context.Context, ) grpc.UnaryServerInterceptor { - const op = "controller.auditResponseInterceptor" + const op = "controller.eventsResponseInterceptor" return func(interceptorCtx context.Context, req any, _ *grpc.UnaryServerInfo, @@ -405,6 +408,9 @@ func auditResponseInterceptor( if err := event.WriteAudit(interceptorCtx, op, event.WithResponse(&event.Response{Details: clonedMsg})); err != nil { return req, status.Errorf(codes.Internal, "unable to write response msg audit: %s", err) } + if err := event.WriteObservation(interceptorCtx, op, event.WithResponse(&event.Response{Details: clonedMsg})); err != nil { + return req, status.Errorf(codes.Internal, "unable to write response msg observation: %s", err) + } } return resp, err diff --git a/internal/daemon/controller/listeners.go b/internal/daemon/controller/listeners.go index b9e1b5137..5c46583dc 100644 --- a/internal/daemon/controller/listeners.go +++ b/internal/daemon/controller/listeners.go @@ -251,8 +251,8 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error grpc.UnaryInterceptor( grpc_middleware.ChainUnaryServer( workerReqInterceptor, - auditRequestInterceptor(c.baseContext), // before we get started, audit the request - auditResponseInterceptor(c.baseContext), // as we finish, audit the response + eventsRequestInterceptor(c.baseContext), // before we get started, send the required events with the request + eventsResponseInterceptor(c.baseContext), // as we finish, send the required events with the response ), ), ) diff --git a/internal/event/cloudevents_formatter_node.go b/internal/event/cloudevents_formatter_node.go index 33090286b..ae436bb21 100644 --- a/internal/event/cloudevents_formatter_node.go +++ b/internal/event/cloudevents_formatter_node.go @@ -69,12 +69,13 @@ func newCloudEventsFormatterFilter(source *url.URL, format cloudevents.Format, o } n.deny = append(n.deny, f) } + } else { + defaultDenyFilters, err := defaultCloudEventsDenyFilters() + if err != nil { + return nil, err + } + n.deny = append(n.deny, defaultDenyFilters...) } - defaultDenyFilters, err := defaultCloudEventsDenyFilters() - if err != nil { - return nil, err - } - n.deny = append(n.deny, defaultDenyFilters...) n.Predicate = newPredicate(n.allow, n.deny) return &n, nil } @@ -83,7 +84,7 @@ func defaultCloudEventsDenyFilters() ([]*filter, error) { const ( op = "event.defaultCloudEventsDenyFilters" // denyWorkStatusEvents is a default filter for worker to controller API status requests - denyWorkStatusEvents = `"/Data/RequestInfo/Method" contains "ServerCoordinationService/Status"` + denyWorkStatusEvents = `"/type" contains "observation" and "/data/request_info/method" contains "ServerCoordinationService/Status"` ) f, err := newFilter(denyWorkStatusEvents) if err != nil { diff --git a/internal/event/cloudevents_formatter_node_test.go b/internal/event/cloudevents_formatter_node_test.go index c43800671..1e7107aba 100644 --- a/internal/event/cloudevents_formatter_node_test.go +++ b/internal/event/cloudevents_formatter_node_test.go @@ -34,9 +34,11 @@ func Test_newCloudEventsFormatterFilter(t *testing.T) { wantDeny []string }{ { - name: "no-opts", - source: testSource, - format: cloudevents.FormatJSON, + // default case should have default deny for filtering ServerCoordinationService/Status for observation events + name: "no-opts", + source: testSource, + format: cloudevents.FormatJSON, + wantDeny: []string{`"/type" contains "observation" and "/data/request_info/method" contains "ServerCoordinationService/Status"`}, }, { name: "bad-allow-filter", @@ -133,7 +135,7 @@ func Test_newCloudEventsFormatterFilter(t *testing.T) { for _, f := range got.allow { assert.Contains(tt.wantAllow, f.raw) } - assert.Len(got.deny, len(tt.wantDeny)+1) // +1 since there's always a default deny + assert.Len(got.deny, len(tt.wantDeny)) defs, err := defaultCloudEventsDenyFilters() require.NoError(err) for _, f := range defs { diff --git a/internal/event/context.go b/internal/event/context.go index 0c1835a0e..792d9ebf9 100644 --- a/internal/event/context.go +++ b/internal/event/context.go @@ -105,8 +105,17 @@ func WriteObservation(ctx context.Context, caller Op, opt ...Option) error { } } opts := getOpts(opt...) - if opts.withDetails == nil && opts.withHeader == nil && !opts.withFlush { - return fmt.Errorf("%s: specify either header or details options for an event payload: %w", op, ErrInvalidParameter) + if opts.withDetails == nil && opts.withHeader == nil && !opts.withFlush && opts.withRequest == nil && opts.withResponse == nil { + return fmt.Errorf("%s: specify either header or details options or request or response for an event payload: "+ + "%w", op, ErrInvalidParameter) + } + // For the case that the telemetry is not enabled, and we have events coming from interceptors. + if !eventer.conf.TelemetryEnabled && (opts.withRequest != nil || opts.withResponse != nil) { + return nil + } + // If telemetry is enabled, we add it to the options. + if eventer.conf.TelemetryEnabled { + opt = append(opt, WithTelemetry()) } if opts.withRequestInfo == nil { var err error diff --git a/internal/event/context_test.go b/internal/event/context_test.go index 71dae61ec..da0b857ef 100644 --- a/internal/event/context_test.go +++ b/internal/event/context_test.go @@ -313,9 +313,12 @@ func Test_WriteObservation(t *testing.T) { name string noOperation bool noFlush bool + telemetryFlag bool observationPayload []observationPayload header map[string]any details map[string]any + Request *event.Request + Response *event.Response ctx context.Context observationSinkFileName string setup func() error @@ -324,9 +327,10 @@ func Test_WriteObservation(t *testing.T) { wantErrContains string }{ { - name: "no-info-event-id", - noFlush: true, - ctx: testCtxNoEventInfoId, + name: "no-info-event-id", + noFlush: true, + telemetryFlag: true, + ctx: testCtxNoEventInfoId, observationPayload: []observationPayload{ { header: []any{"name", "bar"}, @@ -365,6 +369,27 @@ func Test_WriteObservation(t *testing.T) { wantErrIs: event.ErrInvalidParameter, wantErrContains: "specify either header or details options", }, + { + name: "no-header-or-details-in-payload-no-request-no-response", + ctx: testCtx, + noFlush: true, + observationPayload: []observationPayload{ + {}, + }, + wantErrIs: event.ErrInvalidParameter, + wantErrContains: "specify either header or details options or request or response for an event payload", + }, + { + name: "telemetry-not-enabled-but-request-or-response-available", + ctx: testCtx, + noFlush: true, + telemetryFlag: false, + Request: &event.Request{ + Operation: "create-test", + Endpoint: "0.0.0.0", + }, + observationPayload: testPayloads, + }, { name: "no-ctx-eventer-and-syseventer-not-initialized", ctx: context.Background(), @@ -373,9 +398,10 @@ func Test_WriteObservation(t *testing.T) { wantErrContains: "missing both context and system eventer", }, { - name: "use-syseventer", - noFlush: true, - ctx: context.Background(), + name: "use-syseventer", + noFlush: true, + telemetryFlag: true, + ctx: context.Background(), observationPayload: []observationPayload{ { header: []any{"name", "bar"}, @@ -391,8 +417,9 @@ func Test_WriteObservation(t *testing.T) { cleanup: func() { event.TestResetSystEventer(t) }, }, { - name: "use-syseventer-with-cancelled-ctx", - noFlush: true, + name: "use-syseventer-with-cancelled-ctx", + noFlush: true, + telemetryFlag: true, ctx: func() context.Context { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -415,6 +442,7 @@ func Test_WriteObservation(t *testing.T) { { name: "simple", ctx: testCtx, + telemetryFlag: true, observationPayload: testPayloads, header: testWantHeader, details: testWantDetails, @@ -436,7 +464,9 @@ func Test_WriteObservation(t *testing.T) { } require.Greater(len(tt.observationPayload), 0) for _, p := range tt.observationPayload { - err := event.WriteObservation(tt.ctx, event.Op(op), event.WithHeader(p.header...), event.WithDetails(p.details...)) + err := event.WriteObservation(tt.ctx, event.Op(op), event.WithHeader(p.header...), event.WithDetails(p.details...), + event.WithRequest(tt.Request), + event.WithResponse(tt.Response)) if tt.wantErrIs != nil { assert.ErrorIs(err, tt.wantErrIs) if tt.wantErrContains != "" { @@ -450,6 +480,10 @@ func Test_WriteObservation(t *testing.T) { require.NoError(event.WriteObservation(tt.ctx, event.Op(tt.name), event.WithFlush())) } + if !tt.telemetryFlag { + require.Nil(event.WriteObservation(tt.ctx, event.Op(tt.name), event.WithRequest(tt.Request), + event.WithResponse(tt.Response))) + } if tt.observationSinkFileName != "" { defer func() { _ = os.WriteFile(tt.observationSinkFileName, nil, 0o666) }() b, err := ioutil.ReadFile(tt.observationSinkFileName) diff --git a/internal/event/event_observation.go b/internal/event/event_observation.go index efa0daf70..03b074f28 100644 --- a/internal/event/event_observation.go +++ b/internal/event/event_observation.go @@ -100,6 +100,9 @@ func (o *observation) ComposeFrom(events []*eventlogger.Event) (eventlogger.Even payload[hdrK] = hdrV } } + if g.RequestInfo != nil { + payload[RequestInfoField] = g.RequestInfo + } if g.Detail != nil { if _, ok := payload[DetailsField]; !ok { payload[DetailsField] = []gated.EventPayloadDetails{}