(telemetry_filter): handle core proto types atomically

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

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

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: &timestamppb.Timestamp{
+			Seconds: 1694589179,
+			Nanos:   812910000,
+		},
+		TimestampObs: &timestamppb.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, &timestamppb.Timestamp{
+		Seconds: 1694589179,
+		Nanos:   812910000,
+	})
+	assert.Equal(data.WrappedStringObs, &wrapperspb.StringValue{Value: "[REDACTED]"})
+	assert.Equal(data.FieldmaskObs, &fieldmaskpb.FieldMask{
+		Paths: []string{"a", "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",
 		},