(telemetry_filter): handle map / slice traversals of proto structs

internal/event/telemetry_filter.go

@@ -63,8 +63,8 @@ func filterValue(fv reflect.Value, isObservable bool) {
 
 // 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(), ".")
+func packageNameFromType(fieldType reflect.Type) string {
+	typeSegments := strings.Split(fieldType.String(), ".")
 	pkg := strings.TrimLeft(typeSegments[0], "*")
 	return pkg
 }
@@ -89,6 +89,12 @@ func recurseStructureWithProtoFilter(value reflect.Value, filterFunc protoFilter
 				mVal = mVal.Elem()
 				vKind = mVal.Kind()
 			}
+			if coreProtoPackages[packageNameFromType(mVal.Type())] {
+				if !isObservable {
+					value.SetMapIndex(k, reflect.Value{})
+				}
+				continue
+			}
 			switch vKind {
 			case reflect.Struct, reflect.Array, reflect.Slice:
 				if err := recurseStructureWithProtoFilter(mVal, filterFunc, isObservable); err != nil {
@@ -106,19 +112,36 @@ func recurseStructureWithProtoFilter(value reflect.Value, filterFunc protoFilter
 		return nil
 	case reflect.Array, reflect.Slice:
 		if value.Len() > 0 {
+			sliceType := value.Index(0).Type()
+			isCoreProto := coreProtoPackages[packageNameFromType(sliceType)]
 			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 !isCoreProto {
+				for i := 0; i < value.Len(); i++ {
+					sVal := value.Index(i)
+					sKind := sVal.Kind()
+					if sKind == reflect.Ptr || sKind == reflect.Interface {
+						sVal = sVal.Elem()
+						sKind = sVal.Kind()
+					}
+					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 an array is empty, zero its contents
+			if (zeroCount == value.Len() || (isCoreProto && !isObservable)) && value.CanSet() {
+				switch kind {
+				case reflect.Slice:
+					value.SetLen(0)
+				case reflect.Array:
+					for i := 0; i < value.Len(); i++ {
+						value.Index(i).SetZero()
+					}
+				}
 			}
 		}
 	case reflect.Struct:
@@ -134,7 +157,7 @@ func recurseStructureWithProtoFilter(value reflect.Value, filterFunc protoFilter
 				isObservable = filterFunc(field)
 			}
 			// structures in core proto packages are not recursively filtered
-			if coreProtoPackages[packageNameFromType(field)] {
+			if coreProtoPackages[packageNameFromType(field.Type)] {
 				if !isObservable {
 					v.SetZero()
 				}

internal/event/telemetry_filter_test.go

@@ -4,12 +4,15 @@
 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"
 
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/fieldmaskpb"
+	"google.golang.org/protobuf/types/known/structpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+
 	"github.com/hashicorp/boundary/internal/gen/controller/servers"
 	"github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	tassert "github.com/stretchr/testify/assert"
@@ -60,7 +63,7 @@ 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, 1)
+	assert.Len(output.WorkerStatus.Tags, 0)
 	assert.Len(output.Jobs, 1)
 }
 
@@ -198,3 +201,115 @@ func Test_coreProtoTypes(t *testing.T) {
 		Paths: []string{"a", "b"},
 	})
 }
+
+func Test_mapStructPBValues(t *testing.T) {
+	assert := tassert.New(t)
+	type testType struct {
+		ListValueMap    map[string]*structpb.ListValue
+		ListValueMapObs map[string]*structpb.ListValue `eventstream:"observation"`
+	}
+	data := &testType{
+		ListValueMap: map[string]*structpb.ListValue{
+			"one": {Values: []*structpb.Value{
+				structpb.NewStringValue("one"),
+			}},
+			"two": {Values: []*structpb.Value{
+				structpb.NewStringValue("two"),
+			}},
+		},
+		ListValueMapObs: map[string]*structpb.ListValue{
+			"three": {Values: []*structpb.Value{
+				structpb.NewStringValue("three"),
+			}},
+			"four": {Values: []*structpb.Value{
+				structpb.NewStringValue("four"),
+			}},
+		},
+	}
+	err := recurseStructureWithProtoFilter(reflect.ValueOf(data), telemetryFilter, false)
+	assert.NoError(err)
+	assert.Len(data.ListValueMapObs, 2)
+	assert.True(
+		proto.Equal(
+			data.ListValueMapObs["three"],
+			&structpb.ListValue{
+				Values: []*structpb.Value{
+					structpb.NewStringValue("three"),
+				},
+			},
+		),
+	)
+	assert.True(
+		proto.Equal(
+			data.ListValueMapObs["four"],
+			&structpb.ListValue{
+				Values: []*structpb.Value{
+					structpb.NewStringValue("four"),
+				},
+			},
+		),
+	)
+	assert.Len(data.ListValueMap, 0)
+}
+
+func Test_sliceStructPBValues(t *testing.T) {
+	assert := tassert.New(t)
+	type testType struct {
+		ListValueSlice    []*structpb.ListValue
+		ListValueSliceObs []*structpb.ListValue `eventstream:"observation"`
+		ListValueArray    [2]*structpb.ListValue
+	}
+	data := &testType{
+		ListValueSlice: []*structpb.ListValue{
+			{Values: []*structpb.Value{
+				structpb.NewStringValue("one"),
+			}},
+			{Values: []*structpb.Value{
+				structpb.NewStringValue("two"),
+			}},
+		},
+		ListValueSliceObs: []*structpb.ListValue{
+			{Values: []*structpb.Value{
+				structpb.NewStringValue("three"),
+			}},
+			{Values: []*structpb.Value{
+				structpb.NewStringValue("four"),
+			}},
+		},
+		ListValueArray: [2]*structpb.ListValue{
+			{Values: []*structpb.Value{
+				structpb.NewStringValue("five"),
+			}},
+			{Values: []*structpb.Value{
+				structpb.NewStringValue("six"),
+			}},
+		},
+	}
+	err := recurseStructureWithProtoFilter(reflect.ValueOf(data), telemetryFilter, false)
+	assert.NoError(err)
+	assert.Len(data.ListValueSliceObs, 2)
+	assert.True(
+		proto.Equal(
+			data.ListValueSliceObs[0],
+			&structpb.ListValue{
+				Values: []*structpb.Value{
+					structpb.NewStringValue("three"),
+				},
+			},
+		),
+	)
+	assert.True(
+		proto.Equal(
+			data.ListValueSliceObs[1],
+			&structpb.ListValue{
+				Values: []*structpb.Value{
+					structpb.NewStringValue("four"),
+				},
+			},
+		),
+	)
+	assert.Len(data.ListValueSlice, 0)
+	assert.Len(data.ListValueArray, 2)
+	assert.Nil(data.ListValueArray[0])
+	assert.Nil(data.ListValueArray[1])
+}

internal/event/testing.go

@@ -420,7 +420,6 @@ func testWorkerStatusObservable(t testing.TB) *services.StatusRequest {
 		},
 		WorkerStatus: &servers.ServerWorkerStatus{
 			PublicId:         "testID",
-			Tags:             []*servers.TagPair{{}},
 			ReleaseVersion:   "Boundary v0.13.1",
 			OperationalState: "active",
 		},