From c7cd6eabf34c4b523a593495e8e8f4d8fd5e44f5 Mon Sep 17 00:00:00 2001 From: April Ayres Date: Wed, 13 Sep 2023 14:28:31 -0700 Subject: [PATCH] (telemetry_filter): handle core proto types atomically --- internal/event/event_observation_test.go | 8 +- internal/event/telemetry_filter.go | 104 ++++++++++++++++------- internal/event/telemetry_filter_test.go | 99 ++++++++++++++++++++- internal/event/testing.go | 6 +- 4 files changed, 175 insertions(+), 42 deletions(-) diff --git a/internal/event/event_observation_test.go b/internal/event/event_observation_test.go index 97261d45e..6f9d9f245 100644 --- a/internal/event/event_observation_test.go +++ b/internal/event/event_observation_test.go @@ -516,12 +516,8 @@ func Test_composeFromTelemetryFiltering(t *testing.T) { pmsg, ok := req.(*Response) assert.True(ok) assert.NotNil(pmsg) - // fmt.Printf("ACTUAL %+v \n", pmsg.Details) - // j, err := json.Marshal(pmsg) - // assert.NotNil(err) - // fmt.Printf("ACTUAL JSON %s \n", string(j)) - // fmt.Printf("RESPONSE %+v \n", tt.wantFilteredResponse.Details) - // assert.True(proto.Equal(tt.wantFilteredResponse.Details, pmsg.Details)) + assert.True(proto.Equal(tt.wantFilteredResponse.Details, pmsg.Details), "protos differ:\nexpected: %+v\nactual: %+v", tt.wantFilteredResponse.Details, pmsg.Details) + } }) } diff --git a/internal/event/telemetry_filter.go b/internal/event/telemetry_filter.go index 35addc40d..2f32bd141 100644 --- a/internal/event/telemetry_filter.go +++ b/internal/event/telemetry_filter.go @@ -6,17 +6,31 @@ package event import ( "errors" "reflect" + "strings" "google.golang.org/protobuf/proto" ) +// Any fields in these packages should be treated as atomic structures +var coreProtoPackages = map[string]bool{ + "anypb": true, + "apipb": true, + "durationpb": true, + "emptypb": true, + "fieldmaskpb": true, + "sourcecontextpb": true, + "structpb": true, + "timestamppb": true, + "typepb": true, + "wrapperspb": true, +} + // protoFilter is a signature for a struct field validation test type protoFilter func(field reflect.StructField) bool // telemetryFilter checks a struct field should be included in observation telemetry data func telemetryFilter(field reflect.StructField) bool { if field.Tag.Get("eventstream") == "observation" { - // log.Printf("%s is allowed by Tags (%s:\"%s\")", name, tag, tagValue) return true } return false @@ -47,6 +61,17 @@ func filterValue(fv reflect.Value, isObservable bool) { return } +// packageNameFromType extracts the package for a type for use with +// core protobuf type checks. +func packageNameFromType(field reflect.StructField) string { + typeSegments := strings.Split(field.Type.String(), ".") + pkg := strings.TrimLeft(typeSegments[0], "*") + return pkg +} + +// recurseStructureWithProtoFilter will recursively traverse the data struct passed in, and run +// a filter function (filterFunc) on struct fields to decide if they should be preserved or cleared. +// isObservable specifies if the level the current value resides in is considered observable or not. func recurseStructureWithProtoFilter(value reflect.Value, filterFunc protoFilter, isObservable bool) error { kind := value.Kind() @@ -55,44 +80,50 @@ func recurseStructureWithProtoFilter(value reflect.Value, filterFunc protoFilter value = value.Elem() return recurseStructureWithProtoFilter(value, filterFunc, isObservable) case reflect.Map: - // m := reflect.ValueOf(value) - // fmt.Printf("type of m %+v \n", reflect.TypeOf(value)) - for _, k := range value.MapKeys() { - mVal := value.MapIndex(k) - if err := recurseStructureWithProtoFilter(mVal, filterFunc, isObservable); err != nil { - return err + iter := value.MapRange() + for iter.Next() { + mVal := iter.Value() + k := iter.Key() + vKind := mVal.Kind() + if vKind == reflect.Ptr || vKind == reflect.Interface { + mVal = mVal.Elem() + vKind = mVal.Kind() + } + switch vKind { + case reflect.Struct, reflect.Array, reflect.Slice: + if err := recurseStructureWithProtoFilter(mVal, filterFunc, isObservable); err != nil { + return err + } + if mVal.IsValid() && mVal.IsZero() { + value.SetMapIndex(k, reflect.Value{}) + } + default: + if !isObservable { + value.SetMapIndex(k, reflect.Value{}) + } } } return nil case reflect.Array, reflect.Slice: - for i := 0; i < value.Len(); i++ { - sVal := value.Index(i) - // fmt.Printf("Array: sval %+v, is observable %+v \n", sVal, isObservable) - if err := recurseStructureWithProtoFilter(sVal, filterFunc, isObservable); err != nil { - return err + if value.Len() > 0 { + zeroCount := 0 + for i := 0; i < value.Len(); i++ { + sVal := value.Index(i) + if err := recurseStructureWithProtoFilter(sVal, filterFunc, isObservable); err != nil { + return err + } + if sVal.IsValid() && sVal.IsZero() { + zeroCount++ + } + } + // if slice is empty after processing, we can zero its length + if zeroCount == value.Len() && kind == reflect.Slice && value.CanSet() { + value.SetLen(0) } } - //if isObservable { - // for i := 0; i < value.Len(); i++ { - // sVal := value.Index(i) - // if err := recurseStructureWithProtoFilter(sVal, filterFunc, isObservable); err != nil { - // return err - // } - // } - //} else { - // if kind == reflect.Slice { - // value.SetLen(0) // truncate - // } else { - // // fixed size, so we zero - // for i := 0; i < value.Len(); i++ { - // value.Index(i).SetZero() - // } - // } - //} case reflect.Struct: fields := value.Type() - num := fields.NumField() - for i := 0; i < num; i++ { + for i := 0; i < fields.NumField(); i++ { field := fields.Field(i) v := value.Field(i) if !field.IsExported() { @@ -102,7 +133,13 @@ func recurseStructureWithProtoFilter(value reflect.Value, filterFunc protoFilter if filterFunc != nil { isObservable = filterFunc(field) } - // fmt.Printf("field name %+v, is observable %+v \n", field.Name, isObservable) + // structures in core proto packages are not recursively filtered + if coreProtoPackages[packageNameFromType(field)] { + if !isObservable { + v.SetZero() + } + continue + } if err := recurseStructureWithProtoFilter(v, filterFunc, isObservable); err != nil { return err } @@ -117,6 +154,9 @@ func recurseStructureWithProtoFilter(value reflect.Value, filterFunc protoFilter return nil } +// filterProtoMessage recursively traverses the proto.Message passed in, applying the specified filter +// to decide if it will clear or preserve fields. It operates on a clone of the supplied proto, so as to preserve +// the original data. func filterProtoMessage(msg proto.Message, filterFunc protoFilter) (proto.Message, error) { if msg == nil { return nil, errors.New("nil message") diff --git a/internal/event/telemetry_filter_test.go b/internal/event/telemetry_filter_test.go index f31e2423b..e031eac77 100644 --- a/internal/event/telemetry_filter_test.go +++ b/internal/event/telemetry_filter_test.go @@ -4,6 +4,10 @@ package event import ( + "google.golang.org/protobuf/types/known/fieldmaskpb" + "google.golang.org/protobuf/types/known/timestamppb" + "google.golang.org/protobuf/types/known/wrapperspb" + "reflect" "testing" "github.com/hashicorp/boundary/internal/gen/controller/servers" @@ -56,8 +60,8 @@ func Test_OnlyObservationTaggedFieldsPopulated(t *testing.T) { assert.Zero(output.WorkerStatus.Address) assert.Zero(output.WorkerStatus.Description) assert.Zero(output.WorkerStatus.KeyId) - assert.Len(output.WorkerStatus.Tags, 0) - assert.Len(output.Jobs, 0) + assert.Len(output.WorkerStatus.Tags, 1) + assert.Len(output.Jobs, 1) } func Test_AllFieldsPopulatedWithoutFilter(t *testing.T) { @@ -103,3 +107,94 @@ func Test_NilMessageWillError(t *testing.T) { _, err := filterProtoMessage(nil, nil) assert.Error(err) } + +func Test_scalarSliceFilterAllZeroVals(t *testing.T) { + assert := tassert.New(t) + type testType struct { + NonObservableStrings []string + ObservableInts []int `eventstream:"observation"` + } + data := &testType{ + NonObservableStrings: []string{"a", "b", "c"}, + ObservableInts: []int{1, 2, 3, 4}, + } + err := recurseStructureWithProtoFilter(reflect.ValueOf(data), telemetryFilter, false) + assert.NoError(err) + assert.Len(data.NonObservableStrings, 0) + assert.Len(data.ObservableInts, 4) +} + +func Test_mapFilter(t *testing.T) { + assert := tassert.New(t) + type testType struct { + NonObservableMap map[string]int + ObservableMap map[string]int `eventstream:"observation"` + } + data := &testType{ + NonObservableMap: map[string]int{ + "a": 1, + "b": 2, + "c": 3, + }, + ObservableMap: map[string]int{ + "d": 4, + "e": 5, + "f": 6, + }, + } + err := recurseStructureWithProtoFilter(reflect.ValueOf(data), telemetryFilter, false) + assert.NoError(err) + assert.Len(data.NonObservableMap, 0) + assert.Len(data.ObservableMap, 3) + assert.Equal(data.ObservableMap, map[string]int{ + "d": 4, + "e": 5, + "f": 6, + }) +} + +func Test_coreProtoTypes(t *testing.T) { + assert := tassert.New(t) + type testType struct { + Timestamp *timestamppb.Timestamp + TimestampObs *timestamppb.Timestamp `eventstream:"observation"` + WrappedString *wrapperspb.StringValue + WrappedStringObs *wrapperspb.StringValue `eventstream:"observation"` + Fieldmask *fieldmaskpb.FieldMask + FieldmaskObs *fieldmaskpb.FieldMask `eventstream:"observation"` + } + data := &testType{ + Timestamp: ×tamppb.Timestamp{ + Seconds: 1694589179, + Nanos: 812910000, + }, + TimestampObs: ×tamppb.Timestamp{ + Seconds: 1694589179, + Nanos: 812910000, + }, + WrappedString: &wrapperspb.StringValue{Value: "[REDACTED]"}, + WrappedStringObs: &wrapperspb.StringValue{Value: "[REDACTED]"}, + Fieldmask: &fieldmaskpb.FieldMask{ + Paths: []string{"a", "b"}, + }, + FieldmaskObs: &fieldmaskpb.FieldMask{ + Paths: []string{"a", "b"}, + }, + } + err := recurseStructureWithProtoFilter(reflect.ValueOf(data), telemetryFilter, false) + assert.NoError(err) + assert.Nil(data.WrappedString) + assert.Nil(data.Timestamp) + assert.Nil(data.Fieldmask) + assert.NotNil(data.TimestampObs) + assert.NotNil(data.WrappedStringObs) + assert.NotNil(data.FieldmaskObs) + assert.Equal(data.TimestampObs, ×tamppb.Timestamp{ + Seconds: 1694589179, + Nanos: 812910000, + }) + assert.Equal(data.WrappedStringObs, &wrapperspb.StringValue{Value: "[REDACTED]"}) + assert.Equal(data.FieldmaskObs, &fieldmaskpb.FieldMask{ + Paths: []string{"a", "b"}, + }) +} diff --git a/internal/event/testing.go b/internal/event/testing.go index 8ad5bb813..6ea2fa931 100644 --- a/internal/event/testing.go +++ b/internal/event/testing.go @@ -415,10 +415,12 @@ func testWorkerStatus(t testing.TB) *services.StatusRequest { func testWorkerStatusObservable(t testing.TB) *services.StatusRequest { t.Helper() return &services.StatusRequest{ - Jobs: []*services.JobStatus{}, + Jobs: []*services.JobStatus{ + {Job: &services.Job{}}, + }, WorkerStatus: &servers.ServerWorkerStatus{ PublicId: "testID", - Tags: []*servers.TagPair{}, + Tags: []*servers.TagPair{{}}, ReleaseVersion: "Boundary v0.13.1", OperationalState: "active", },