Fix agent cache incorrectly notifying unchanged protobufs. (#15866)

Fix agent cache incorrectly notifying unchanged protobufs.

This change fixes a situation where the protobuf private fields
would be read by reflect.DeepEqual() and indicate data was modified.
This resulted in change notifications being fired every time, which
could cause performance problems in proxycfg.

.changelog/15866.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+agent: Fix issue where the agent cache would incorrectly mark protobuf objects as updated.
+```

agent/cache/watch.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/consul/lib"
+	"google.golang.org/protobuf/proto"
 )
 
 // UpdateEvent is a struct summarizing an update to a cache entry
@@ -181,7 +182,7 @@ func (c *Cache) notifyPollingQuery(ctx context.Context, r getOptions, correlatio
 		}
 
 		// Check for a change in the value or an index change
-		if index < meta.Index || !reflect.DeepEqual(lastValue, res) {
+		if index < meta.Index || !isEqual(lastValue, res) {
 			cb(ctx, UpdateEvent{correlationID, res, meta, err})
 
 			// Update index and lastValue
@@ -251,3 +252,20 @@ func (c *Cache) notifyPollingQuery(ctx context.Context, r getOptions, correlatio
 		}
 	}
 }
+
+// isEqual compares two values for deep equality. Protobuf objects
+// require us to use a special comparison because cloned structs
+// may have non-exported fields that differ. For non-protobuf objects,
+// we use reflect.DeepEqual().
+func isEqual(a, b interface{}) bool {
+	// TODO move this logic into an interface so that each type can determine
+	// its own logic for equality, rather than a centralized type-cast like this.
+	if a != nil && b != nil {
+		a, aok := a.(proto.Message)
+		b, bok := b.(proto.Message)
+		if aok && bok {
+			return proto.Equal(a, b)
+		}
+	}
+	return reflect.DeepEqual(a, b)
+}

agent/cache/watch_test.go

@@ -4,12 +4,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"reflect"
 	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 // Test that a type registered with a periodic refresh can be watched.
@@ -401,3 +404,44 @@ OUT:
 	actual := atomic.LoadUint32(&retries)
 	require.True(t, actual < 10, fmt.Sprintf("actual: %d", actual))
 }
+
+func Test_isEqual(t *testing.T) {
+	s1 := time.Now()
+	s2 := s1
+
+	var pb1 proto.Message = timestamppb.New(s1)
+	var pb2 proto.Message = &timestamppb.Timestamp{}
+
+	// Marshal and unmarshal the proto object
+	b, err := proto.Marshal(pb1)
+	require.NoError(t, err)
+	require.NoError(t, proto.Unmarshal(b, pb2))
+
+	// Sanity-check that protobuf comparisons fail reflect.DeepEqual
+	require.False(t, reflect.DeepEqual(pb1, pb2))
+
+	tests := []struct {
+		equal bool
+		v1    interface{}
+		v2    interface{}
+	}{
+		// Test protobuf message comparisons work.
+		{true, nil, nil},
+		{true, pb1, pb2},
+		{false, nil, pb2},
+		{false, pb1, nil},
+		{false, pb1, timestamppb.New(s1.Add(time.Second))},
+		// Test normal struct comparisons
+		{true, s1, s2},
+		{true, &s1, &s2},
+		{false, s1, nil},
+		{false, &s1, nil},
+		{false, nil, s2},
+		{false, nil, &s2},
+		{false, s1, s1.Add(time.Second)},
+	}
+
+	for _, tc := range tests {
+		require.Equal(t, tc.equal, isEqual(tc.v1, tc.v2))
+	}
+}