From 80a448984424ac5d8590f8fb890b1b26bca3186d Mon Sep 17 00:00:00 2001 From: Kyle Havlovitz Date: Fri, 10 Dec 2021 13:36:10 -0800 Subject: [PATCH] state: fix freed VIP table id index --- agent/consul/state/catalog_events_test.go | 17 ++- agent/consul/state/catalog_schema.go | 25 ++-- agent/consul/state/catalog_test.go | 139 +++++++++++++++++++++- 3 files changed, 170 insertions(+), 11 deletions(-) diff --git a/agent/consul/state/catalog_events_test.go b/agent/consul/state/catalog_events_test.go index 6e86171d21..41d5a0961a 100644 --- a/agent/consul/state/catalog_events_test.go +++ b/agent/consul/state/catalog_events_test.go @@ -1033,8 +1033,8 @@ func TestServiceHealthEventsFromChanges(t *testing.T) { testServiceHealthEvent(t, "web", evSidecar, evNodeUnchanged), testServiceHealthEvent(t, "web", evConnectTopic, evSidecar, evNodeUnchanged), - testServiceHealthEvent(t, "api", evNode2, evConnectNative, evNodeUnchanged), - testServiceHealthEvent(t, "api", evNode2, evConnectTopic, evConnectNative, evNodeUnchanged), + testServiceHealthEvent(t, "api", evNode2, evConnectNative, evNodeUnchanged, evVirtualIPChanged("240.0.0.2")), + testServiceHealthEvent(t, "api", evNode2, evConnectTopic, evConnectNative, evNodeUnchanged, evVirtualIPChanged("240.0.0.2")), }, }) run(t, eventsTestCase{ @@ -2116,6 +2116,19 @@ func evTerminatingGatewayRenamed(newName string) func(e *stream.Event) error { } } +func evVirtualIPChanged(newIP string) func(e *stream.Event) error { + return func(e *stream.Event) error { + csn := getPayloadCheckServiceNode(e.Payload) + csn.Service.TaggedAddresses = map[string]structs.ServiceAddress{ + structs.TaggedAddressVirtualIP: { + Address: newIP, + Port: csn.Service.Port, + }, + } + return nil + } +} + // evNodeMeta option alters the base event node to add some meta data. func evNodeMeta(e *stream.Event) error { csn := getPayloadCheckServiceNode(e.Payload) diff --git a/agent/consul/state/catalog_schema.go b/agent/consul/state/catalog_schema.go index c03f649be8..9d0b447dc0 100644 --- a/agent/consul/state/catalog_schema.go +++ b/agent/consul/state/catalog_schema.go @@ -618,6 +618,13 @@ type FreeVirtualIP struct { IsCounter bool } +func counterIndex(obj interface{}) (bool, error) { + if vip, ok := obj.(FreeVirtualIP); ok { + return vip.IsCounter, nil + } + return false, fmt.Errorf("object is not a virtual IP entry") +} + func serviceVirtualIPTableSchema() *memdb.TableSchema { return &memdb.TableSchema{ Name: tableServiceVirtualIPs, @@ -642,8 +649,15 @@ func freeVirtualIPTableSchema() *memdb.TableSchema { Name: indexID, AllowMissing: false, Unique: true, - Indexer: &memdb.StringFieldIndex{ - Field: "IP", + Indexer: &memdb.CompoundIndex{ + Indexes: []memdb.Indexer{ + &memdb.StringFieldIndex{ + Field: "IP", + }, + &memdb.ConditionalIndex{ + Conditional: counterIndex, + }, + }, }, }, indexCounterOnly: { @@ -651,12 +665,7 @@ func freeVirtualIPTableSchema() *memdb.TableSchema { AllowMissing: false, Unique: false, Indexer: &memdb.ConditionalIndex{ - Conditional: func(obj interface{}) (bool, error) { - if vip, ok := obj.(FreeVirtualIP); ok { - return vip.IsCounter, nil - } - return false, fmt.Errorf("object is not a virtual IP entry") - }, + Conditional: counterIndex, }, }, }, diff --git a/agent/consul/state/catalog_test.go b/agent/consul/state/catalog_test.go index bfca9a2d9a..1071fea23f 100644 --- a/agent/consul/state/catalog_test.go +++ b/agent/consul/state/catalog_test.go @@ -1548,7 +1548,7 @@ func TestStateStore_EnsureService_connectProxy(t *testing.T) { assert.Equal(&expect1, out.Services["connect-proxy"]) } -func TestStateStore_EnsureService_virtualIps(t *testing.T) { +func TestStateStore_EnsureService_VirtualIPAssign(t *testing.T) { assert := assert.New(t) s := testStateStore(t) require.NoError(t, s.SystemMetadataSet(0, &structs.SystemMetadataEntry{ @@ -1684,6 +1684,143 @@ func TestStateStore_EnsureService_virtualIps(t *testing.T) { assert.Equal(ns4.Port, taggedAddress.Port) } +func TestStateStore_EnsureService_ReassignFreedVIPs(t *testing.T) { + assert := assert.New(t) + s := testStateStore(t) + require.NoError(t, s.SystemMetadataSet(0, &structs.SystemMetadataEntry{ + Key: structs.SystemMetadataVirtualIPsEnabled, + Value: "true", + })) + + // Create the service registration. + entMeta := structs.DefaultEnterpriseMetaInDefaultPartition() + ns1 := &structs.NodeService{ + ID: "foo", + Service: "foo", + Address: "1.1.1.1", + Port: 1111, + Weights: &structs.Weights{ + Passing: 1, + Warning: 1, + }, + Connect: structs.ServiceConnect{Native: true}, + EnterpriseMeta: *entMeta, + } + + // Service successfully registers into the state store. + testRegisterNode(t, s, 0, "node1") + require.NoError(t, s.EnsureService(10, "node1", ns1)) + + // Make sure there's a virtual IP for the foo service. + vip, err := s.VirtualIPForService(structs.ServiceName{Name: "foo"}) + require.NoError(t, err) + assert.Equal("240.0.0.1", vip) + + // Retrieve and verify + _, out, err := s.NodeServices(nil, "node1", nil) + require.NoError(t, err) + assert.NotNil(out) + + taggedAddress := out.Services["foo"].TaggedAddresses[structs.TaggedAddressVirtualIP] + assert.Equal(vip, taggedAddress.Address) + assert.Equal(ns1.Port, taggedAddress.Port) + + // Create the service registration. + ns2 := &structs.NodeService{ + Kind: structs.ServiceKindTypical, + ID: "redis", + Service: "redis", + Address: "2.2.2.2", + Port: 2222, + Weights: &structs.Weights{ + Passing: 1, + Warning: 1, + }, + Connect: structs.ServiceConnect{Native: true}, + EnterpriseMeta: *entMeta, + } + require.NoError(t, s.EnsureService(11, "node1", ns2)) + + // Make sure the virtual IP has been incremented for the redis service. + vip, err = s.VirtualIPForService(structs.ServiceName{Name: "redis"}) + require.NoError(t, err) + assert.Equal("240.0.0.2", vip) + + // Retrieve and verify + _, out, err = s.NodeServices(nil, "node1", nil) + assert.Nil(err) + assert.NotNil(out) + + taggedAddress = out.Services["redis"].TaggedAddresses[structs.TaggedAddressVirtualIP] + assert.Equal(vip, taggedAddress.Address) + assert.Equal(ns2.Port, taggedAddress.Port) + + // Delete the last service and make sure it no longer has a virtual IP assigned. + require.NoError(t, s.DeleteService(12, "node1", "redis", entMeta)) + vip, err = s.VirtualIPForService(structs.ServiceName{Name: "redis"}) + require.NoError(t, err) + assert.Equal("", vip) + + // Register a new service, should end up with the freed 240.0.0.2 address. + ns3 := &structs.NodeService{ + Kind: structs.ServiceKindTypical, + ID: "backend", + Service: "backend", + Address: "2.2.2.2", + Port: 2222, + Weights: &structs.Weights{ + Passing: 1, + Warning: 1, + }, + Connect: structs.ServiceConnect{Native: true}, + EnterpriseMeta: *entMeta, + } + require.NoError(t, s.EnsureService(13, "node1", ns3)) + + vip, err = s.VirtualIPForService(structs.ServiceName{Name: "backend"}) + require.NoError(t, err) + assert.Equal("240.0.0.2", vip) + + // Retrieve and verify + _, out, err = s.NodeServices(nil, "node1", nil) + assert.Nil(err) + assert.NotNil(out) + + taggedAddress = out.Services["backend"].TaggedAddresses[structs.TaggedAddressVirtualIP] + assert.Equal(vip, taggedAddress.Address) + assert.Equal(ns3.Port, taggedAddress.Port) + + // Create a new service, no more freed VIPs so it should go back to using the counter. + ns4 := &structs.NodeService{ + Kind: structs.ServiceKindTypical, + ID: "frontend", + Service: "frontend", + Address: "2.2.2.2", + Port: 2222, + Weights: &structs.Weights{ + Passing: 1, + Warning: 1, + }, + Connect: structs.ServiceConnect{Native: true}, + EnterpriseMeta: *entMeta, + } + require.NoError(t, s.EnsureService(14, "node1", ns4)) + + // Make sure the virtual IP has been incremented for the frontend service. + vip, err = s.VirtualIPForService(structs.ServiceName{Name: "frontend"}) + require.NoError(t, err) + assert.Equal("240.0.0.3", vip) + + // Retrieve and verify + _, out, err = s.NodeServices(nil, "node1", nil) + assert.Nil(err) + assert.NotNil(out) + + taggedAddress = out.Services["frontend"].TaggedAddresses[structs.TaggedAddressVirtualIP] + assert.Equal(vip, taggedAddress.Address) + assert.Equal(ns4.Port, taggedAddress.Port) +} + func TestStateStore_Services(t *testing.T) { s := testStateStore(t)