Hash based config entry replication (#19795)

* add a hash to config entries when normalizing

* add GetHash and implement comparing hashes

* only update if the Hash is different

* only update if the Hash is different and not 0

* fix proto to include the Hash

* fix proto gen

* buf format

* add SetHash and fix tests

* fix config load tests

* fix state test and config test

* recalculate hash when restoring config entries

* fix snapshot restore test

* add changelog

* fix missing normalize, fix proto indexes and add normalize test

.changelog/19795.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+wan-federation: use a hash to diff config entries when replicating in the secondary DC to avoid unnecessary writes..
+```

agent/config/runtime_test.go

@@ -6227,6 +6227,9 @@ func (tc testCase) run(format string, dataDir string) func(t *testing.T) {
 		expected.ACLResolverSettings.NodeName = expected.NodeName
 		expected.ACLResolverSettings.EnterpriseMeta = *structs.NodeEnterpriseMetaInPartition(expected.PartitionOrDefault())
 
+		for i, e := range expected.ConfigEntryBootstrap {
+			e.SetHash(actual.ConfigEntryBootstrap[i].GetHash())
+		}
 		prototest.AssertDeepEqual(t, expected, actual, cmpopts.EquateEmpty())
 		if tc.cleanup != nil {
 			tc.cleanup()
@@ -7122,6 +7125,9 @@ func TestLoad_FullConfig(t *testing.T) {
 				time.Date(2019, 11, 20, 5, 0, 0, 0, time.UTC)))
 			r, err := Load(opts)
 			require.NoError(t, err)
+			for i, e := range expected.ConfigEntryBootstrap {
+				e.SetHash(r.RuntimeConfig.ConfigEntryBootstrap[i].GetHash())
+			}
 			prototest.AssertDeepEqual(t, expected, r.RuntimeConfig)
 			require.ElementsMatch(t, expectedWarns, r.Warnings, "Warnings: %#v", r.Warnings)
 		})

agent/configentry/compare.go

@@ -38,3 +38,10 @@ func EqualID(e1, e2 structs.ConfigEntry) bool {
 		e1.GetEnterpriseMeta().IsSame(e2.GetEnterpriseMeta()) &&
 		e1.GetName() == e2.GetName()
 }
+
+func SameHash(e1, e2 structs.ConfigEntry) bool {
+	if e1.GetHash() == 0 || e2.GetHash() == 0 {
+		return false
+	}
+	return e1.GetHash() == e2.GetHash()
+}

agent/consul/config_replication.go

@@ -30,7 +30,9 @@ func diffConfigEntries(local []structs.ConfigEntry, remote []structs.ConfigEntry
 		if configentry.EqualID(local[localIdx], remote[remoteIdx]) {
 			// config is in both the local and remote state - need to check raft indices
 			if remote[remoteIdx].GetRaftIndex().ModifyIndex > lastRemoteIndex {
-				updates = append(updates, remote[remoteIdx])
+				if !configentry.SameHash(local[localIdx], remote[remoteIdx]) {
+					updates = append(updates, remote[remoteIdx])
+				}
 			}
 			// increment both indices when equal
 			localIdx += 1

agent/consul/config_replication_test.go

@@ -6,6 +6,8 @@ package consul
 import (
 	"context"
 	"fmt"
+	"github.com/oklog/ulid/v2"
+	"github.com/stretchr/testify/assert"
 	"os"
 	"testing"
 
@@ -268,3 +270,63 @@ func TestReplication_ConfigEntries_GraphValidationErrorDuringReplication(t *test
 		checkSame(r)
 	})
 }
+
+func createConfigEntries(num int, indexStart int) []structs.ConfigEntry {
+	entries := make([]structs.ConfigEntry, num)
+	for i := range entries {
+		entries[i] = &structs.ServiceConfigEntry{Name: ulid.Make().String(), RaftIndex: structs.RaftIndex{ModifyIndex: uint64(i + indexStart)}}
+	}
+	return entries
+}
+
+func mutateIDs(e []structs.ConfigEntry, indexStart int) []structs.ConfigEntry {
+	entries := make([]structs.ConfigEntry, len(e))
+	for i := range entries {
+		entries[i] = &structs.ServiceConfigEntry{Name: e[i].GetName(), RaftIndex: structs.RaftIndex{ModifyIndex: uint64(i + indexStart)}}
+	}
+	return entries
+}
+
+func Test_diffConfigEntries(t *testing.T) {
+	type args struct {
+		local           []structs.ConfigEntry
+		remote          []structs.ConfigEntry
+		lastRemoteIndex uint64
+		normalize       bool
+	}
+
+	entries1 := createConfigEntries(10, 10)
+	entries2 := createConfigEntries(10, 20)
+	entries3 := append(entries1, entries2...)
+	entries4 := mutateIDs(entries1, 20)
+	entries5 := mutateIDs(entries1, 0)
+	tests := []struct {
+		name    string
+		args    args
+		updated []structs.ConfigEntry
+		deleted []structs.ConfigEntry
+	}{
+		{"empty", args{local: make([]structs.ConfigEntry, 0), remote: make([]structs.ConfigEntry, 0), lastRemoteIndex: 0, normalize: true}, nil, nil},
+		{"same", args{local: entries1, remote: entries1, lastRemoteIndex: 0, normalize: true}, nil, nil},
+		{"new remote", args{local: nil, remote: entries1, lastRemoteIndex: 0, normalize: true}, entries1, nil},
+		{"extra remote", args{local: entries1, remote: entries3, lastRemoteIndex: 0, normalize: true}, entries2, nil},
+		{"extra local", args{local: entries3, remote: entries1, lastRemoteIndex: 0, normalize: true}, nil, entries2},
+		{"same, same size, different raft ID", args{local: entries1, remote: entries4, lastRemoteIndex: 0, normalize: true}, nil, nil},
+		{"when hash is empty, avoid hash compare", args{local: entries5, remote: entries4, lastRemoteIndex: 0, normalize: false}, entries4, nil},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if tt.args.normalize {
+				for _, l := range tt.args.local {
+					require.NoError(t, l.Normalize())
+				}
+				for _, r := range tt.args.remote {
+					require.NoError(t, r.Normalize())
+				}
+			}
+			deletions, updates := diffConfigEntries(tt.args.local, tt.args.remote, tt.args.lastRemoteIndex)
+			assert.Equalf(t, tt.updated, updates, "updated diffConfigEntries(%v, %v, %v)", tt.args.local, tt.args.remote, tt.args.lastRemoteIndex)
+			assert.Equalf(t, tt.deleted, deletions, "deleted diffConfigEntries(%v, %v, %v)", tt.args.local, tt.args.remote, tt.args.lastRemoteIndex)
+		})
+	}
+}

agent/consul/fsm/snapshot_test.go

@@ -815,14 +815,17 @@ func TestFSM_SnapshotRestore_CE(t *testing.T) {
 	// Verify config entries are restored
 	_, serviceConfEntry, err := fsm2.state.ConfigEntry(nil, structs.ServiceDefaults, "foo", structs.DefaultEnterpriseMetaInDefaultPartition())
 	require.NoError(t, err)
+	serviceConfig.SetHash(serviceConfEntry.GetHash())
 	require.Equal(t, serviceConfig, serviceConfEntry)
 
 	_, proxyConfEntry, err := fsm2.state.ConfigEntry(nil, structs.ProxyDefaults, "global", structs.DefaultEnterpriseMetaInDefaultPartition())
 	require.NoError(t, err)
+	proxyConfig.SetHash(proxyConfEntry.GetHash())
 	require.Equal(t, proxyConfig, proxyConfEntry)
 
 	_, ingressRestored, err := fsm2.state.ConfigEntry(nil, structs.IngressGateway, "ingress", structs.DefaultEnterpriseMetaInDefaultPartition())
 	require.NoError(t, err)
+	ingress.SetHash(ingressRestored.GetHash())
 	require.Equal(t, ingress, ingressRestored)
 
 	_, restoredGatewayServices, err := fsm2.state.GatewayServices(nil, "ingress", structs.DefaultEnterpriseMetaInDefaultPartition())
@@ -856,11 +859,13 @@ func TestFSM_SnapshotRestore_CE(t *testing.T) {
 	// Verify service-intentions is restored
 	_, serviceIxnEntry, err := fsm2.state.ConfigEntry(nil, structs.ServiceIntentions, "foo", structs.DefaultEnterpriseMetaInDefaultPartition())
 	require.NoError(t, err)
+	serviceIxn.SetHash(serviceIxnEntry.GetHash())
 	require.Equal(t, serviceIxn, serviceIxnEntry)
 
 	// Verify mesh config entry is restored
 	_, meshConfigEntry, err := fsm2.state.ConfigEntry(nil, structs.MeshConfig, structs.MeshConfigMesh, structs.DefaultEnterpriseMetaInDefaultPartition())
 	require.NoError(t, err)
+	meshConfig.SetHash(meshConfigEntry.GetHash())
 	require.Equal(t, meshConfig, meshConfigEntry)
 
 	_, restoredServiceNames, err := fsm2.state.ServiceNamesOfKind(nil, structs.ServiceKindTypical)

agent/consul/state/config_entry.go

@@ -103,6 +103,13 @@ func (s *Snapshot) ConfigEntries() ([]structs.ConfigEntry, error) {
 
 // ConfigEntry is used when restoring from a snapshot.
 func (s *Restore) ConfigEntry(c structs.ConfigEntry) error {
+	// the hash is recalculated when restoring config entries
+	// in case a new field is added in a newer version.
+	h, err := structs.HashConfigEntry(c)
+	if err != nil {
+		return err
+	}
+	c.SetHash(h)
 	return insertConfigEntryWithTxn(s.tx, c.GetRaftIndex().ModifyIndex, c)
 }
 

agent/consul/state/intention_test.go

@@ -313,6 +313,7 @@ func testStore_IntentionMutation(t *testing.T, s *Store) {
 				src.LegacyMeta = nil
 			}
 		}
+		expect.SetHash(got.GetHash())
 		require.Equal(t, expect, got)
 	}
 

agent/structs/config_entry.go

@@ -95,6 +95,16 @@ type ConfigEntry interface {
 	GetMeta() map[string]string
 	GetEnterpriseMeta() *acl.EnterpriseMeta
 	GetRaftIndex() *RaftIndex
+	GetHash() uint64
+	SetHash(h uint64)
+}
+
+func HashConfigEntry(conf ConfigEntry) (uint64, error) {
+	hash, err := hashstructure.Hash(conf, nil)
+	if err != nil {
+		return hash, err
+	}
+	return hash, nil
 }
 
 // ControlledConfigEntry is an optional interface implemented by a ConfigEntry
@@ -169,8 +179,17 @@ type ServiceConfigEntry struct {
 	EnvoyExtensions           EnvoyExtensions        `json:",omitempty" alias:"envoy_extensions"`
 
 	Meta               map[string]string `json:",omitempty"`
+	Hash               uint64            `json:",omitempty" hash:"ignore"`
 	acl.EnterpriseMeta `hcl:",squash" mapstructure:",squash"`
-	RaftIndex
+	RaftIndex          `hash:"ignore"`
+}
+
+func (e *ServiceConfigEntry) SetHash(h uint64) {
+	e.Hash = h
+}
+
+func (e *ServiceConfigEntry) GetHash() uint64 {
+	return e.Hash
 }
 
 func (e *ServiceConfigEntry) Clone() *ServiceConfigEntry {
@@ -225,6 +244,11 @@ func (e *ServiceConfigEntry) Normalize() error {
 			}
 		}
 	}
+	h, err := HashConfigEntry(e)
+	if err != nil {
+		return err
+	}
+	e.Hash = h
 
 	return validationErr
 }
@@ -451,8 +475,17 @@ type ProxyConfigEntry struct {
 	PrioritizeByLocality *ServiceResolverPrioritizeByLocality `json:",omitempty" alias:"prioritize_by_locality"`
 
 	Meta               map[string]string `json:",omitempty"`
+	Hash               uint64            `json:",omitempty" hash:"ignore"`
 	acl.EnterpriseMeta `hcl:",squash" mapstructure:",squash"`
-	RaftIndex
+	RaftIndex          `hash:"ignore"`
+}
+
+func (e *ProxyConfigEntry) SetHash(h uint64) {
+	e.Hash = h
+}
+
+func (e *ProxyConfigEntry) GetHash() uint64 {
+	return e.Hash
 }
 
 func (e *ProxyConfigEntry) GetKind() string {
@@ -491,7 +524,13 @@ func (e *ProxyConfigEntry) Normalize() error {
 
 	e.EnterpriseMeta.Normalize()
 
+	h, err := HashConfigEntry(e)
+	if err != nil {
+		return err
+	}
+	e.Hash = h
 	return nil
+
 }
 
 func (e *ProxyConfigEntry) Validate() error {

agent/structs/config_entry_discoverychain.go

@@ -79,8 +79,17 @@ type ServiceRouterConfigEntry struct {
 	Routes []ServiceRoute
 
 	Meta               map[string]string `json:",omitempty"`
+	Hash               uint64            `json:",omitempty" hash:"ignore"`
 	acl.EnterpriseMeta `hcl:",squash" mapstructure:",squash"`
-	RaftIndex
+	RaftIndex          `hash:"ignore"`
+}
+
+func (e *ServiceRouterConfigEntry) SetHash(h uint64) {
+	e.Hash = h
+}
+
+func (e *ServiceRouterConfigEntry) GetHash() uint64 {
+	return e.Hash
 }
 
 func (e *ServiceRouterConfigEntry) GetKind() string {
@@ -129,6 +138,11 @@ func (e *ServiceRouterConfigEntry) Normalize() error {
 		}
 	}
 
+	h, err := HashConfigEntry(e)
+	if err != nil {
+		return err
+	}
+	e.Hash = h
 	return nil
 }
 
@@ -537,8 +551,17 @@ type ServiceSplitterConfigEntry struct {
 	Splits []ServiceSplit
 
 	Meta               map[string]string `json:",omitempty"`
+	Hash               uint64            `json:",omitempty" hash:"ignore"`
 	acl.EnterpriseMeta `hcl:",squash" mapstructure:",squash"`
-	RaftIndex
+	RaftIndex          `hash:"ignore"`
+}
+
+func (e *ServiceSplitterConfigEntry) SetHash(h uint64) {
+	e.Hash = h
+}
+
+func (e *ServiceSplitterConfigEntry) GetHash() uint64 {
+	return e.Hash
 }
 
 func (e *ServiceSplitterConfigEntry) GetKind() string {
@@ -581,6 +604,11 @@ func (e *ServiceSplitterConfigEntry) Normalize() error {
 		}
 	}
 
+	h, err := HashConfigEntry(e)
+	if err != nil {
+		return err
+	}
+	e.Hash = h
 	return nil
 }
 
@@ -876,8 +904,17 @@ type ServiceResolverConfigEntry struct {
 	LoadBalancer *LoadBalancer `json:",omitempty" alias:"load_balancer"`
 
 	Meta               map[string]string `json:",omitempty"`
+	Hash               uint64            `json:",omitempty" hash:"ignore"`
 	acl.EnterpriseMeta `hcl:",squash" mapstructure:",squash"`
-	RaftIndex
+	RaftIndex          `hash:"ignore"`
+}
+
+func (e *ServiceResolverConfigEntry) SetHash(h uint64) {
+	e.Hash = h
+}
+
+func (e *ServiceResolverConfigEntry) GetHash() uint64 {
+	return e.Hash
 }
 
 func (e *ServiceResolverConfigEntry) RelatedPeers() []string {
@@ -998,6 +1035,11 @@ func (e *ServiceResolverConfigEntry) Normalize() error {
 
 	e.EnterpriseMeta.Normalize()
 
+	h, err := HashConfigEntry(e)
+	if err != nil {
+		return err
+	}
+	e.Hash = h
 	return nil
 }
 

agent/structs/config_entry_exports.go

@@ -20,8 +20,17 @@ type ExportedServicesConfigEntry struct {
 	Services []ExportedService `json:",omitempty"`
 
 	Meta               map[string]string `json:",omitempty"`
+	Hash               uint64            `json:",omitempty" hash:"ignore"`
 	acl.EnterpriseMeta `hcl:",squash" mapstructure:",squash"`
-	RaftIndex
+	RaftIndex          `hash:"ignore"`
+}
+
+func (e *ExportedServicesConfigEntry) SetHash(h uint64) {
+	e.Hash = h
+}
+
+func (e *ExportedServicesConfigEntry) GetHash() uint64 {
+	return e.Hash
 }
 
 // ExportedService manages the exporting of a service in the local partition to
@@ -79,6 +88,11 @@ func (e *ExportedServicesConfigEntry) Normalize() error {
 	for i := range e.Services {
 		e.Services[i].Namespace = acl.NormalizeNamespace(e.Services[i].Namespace)
 	}
+	h, err := HashConfigEntry(e)
+	if err != nil {
+		return err
+	}
+	e.Hash = h
 
 	return nil
 }