From 38540e93e78e860e92b106741dc2db44ad554e1c Mon Sep 17 00:00:00 2001 From: Yanjun Zhou Date: Fri, 17 Jan 2025 11:30:55 +0800 Subject: [PATCH] Fix Subnet update error detection (#991) Operator modifies the existing Subnet in store before updating the NSX Subnet, which results in the next reconcile will find the updated Subnet already in store, skip the update, and overwrite the Status as Ready. This PR uses a copy when updating the Subnet to avoid this issue. Signed-off-by: Yanjun Zhou --- pkg/nsx/services/subnet/builder.go | 9 +- pkg/nsx/services/subnet/subnet.go | 43 +++---- pkg/nsx/services/subnet/subnet_test.go | 149 +++++++++++++++++++++++++ 3 files changed, 171 insertions(+), 30 deletions(-) diff --git a/pkg/nsx/services/subnet/builder.go b/pkg/nsx/services/subnet/builder.go index 39d995976..6a0f6e3ff 100644 --- a/pkg/nsx/services/subnet/builder.go +++ b/pkg/nsx/services/subnet/builder.go @@ -70,7 +70,7 @@ func (service *SubnetService) buildSubnet(obj client.Object, tags []model.Tag) ( if dhcpMode == "" { dhcpMode = v1alpha1.DHCPConfigModeDeactivated } - nsxSubnet.SubnetDhcpConfig = service.buildSubnetDHCPConfig(dhcpMode) + nsxSubnet.SubnetDhcpConfig = service.buildSubnetDHCPConfig(dhcpMode, nil) nsxSubnet.IpAddresses = o.Spec.IPAddresses case *v1alpha1.SubnetSet: // The index is a random string with the length of 8 chars. It is the first 8 chars of the hash @@ -87,7 +87,7 @@ func (service *SubnetService) buildSubnet(obj client.Object, tags []model.Tag) ( if dhcpMode == "" { dhcpMode = v1alpha1.DHCPConfigModeDeactivated } - nsxSubnet.SubnetDhcpConfig = service.buildSubnetDHCPConfig(dhcpMode) + nsxSubnet.SubnetDhcpConfig = service.buildSubnetDHCPConfig(dhcpMode, nil) default: return nil, SubnetTypeError } @@ -105,10 +105,11 @@ func (service *SubnetService) buildSubnet(obj client.Object, tags []model.Tag) ( return nsxSubnet, nil } -func (service *SubnetService) buildSubnetDHCPConfig(mode string) *model.SubnetDhcpConfig { +func (service *SubnetService) buildSubnetDHCPConfig(mode string, dhcpServerAdditionalConfig *model.DhcpServerAdditionalConfig) *model.SubnetDhcpConfig { nsxMode := nsxutil.ParseDHCPMode(mode) subnetDhcpConfig := &model.SubnetDhcpConfig{ - Mode: &nsxMode, + DhcpServerAdditionalConfig: dhcpServerAdditionalConfig, + Mode: &nsxMode, } return subnetDhcpConfig } diff --git a/pkg/nsx/services/subnet/subnet.go b/pkg/nsx/services/subnet/subnet.go index 40dff1378..04df49b1c 100644 --- a/pkg/nsx/services/subnet/subnet.go +++ b/pkg/nsx/services/subnet/subnet.go @@ -99,12 +99,12 @@ func (service *SubnetService) CreateOrUpdateSubnet(obj client.Object, vpcInfo co if changed { // Only tags and dhcp are expected to be updated // inherit other fields from the existing Subnet - existingSubnet.Tags = nsxSubnet.Tags - if existingSubnet.SubnetDhcpConfig != nil { - existingSubnet.SubnetDhcpConfig = nsxSubnet.SubnetDhcpConfig - existingSubnet.AdvancedConfig.StaticIpAllocation.Enabled = nsxSubnet.AdvancedConfig.StaticIpAllocation.Enabled - } - nsxSubnet = existingSubnet + // Avoid modification on existingSubnet to ensure + // Subnet store is only updated after the updating succeeds. + updatedSubnet := *existingSubnet + updatedSubnet.Tags = nsxSubnet.Tags + updatedSubnet.SubnetDhcpConfig = nsxSubnet.SubnetDhcpConfig + nsxSubnet = &updatedSubnet } } if !changed { @@ -391,7 +391,6 @@ func (service *SubnetService) UpdateSubnetSet(ns string, vpcSubnets []*model.Vpc if dhcpMode == "" { dhcpMode = v1alpha1.DHCPConfigModeDeactivated } - staticIpAllocation := dhcpMode == v1alpha1.DHCPConfigModeDeactivated for i, vpcSubnet := range vpcSubnets { subnetSet := &v1alpha1.SubnetSet{} var name string @@ -421,36 +420,28 @@ func (service *SubnetService) UpdateSubnetSet(ns string, vpcSubnets []*model.Vpc } newTags := append(service.buildBasicTags(subnetSet), tags...) - var updatedSubnet *model.VpcSubnet - if vpcSubnets[i].SubnetDhcpConfig == nil { - updatedSubnet = &model.VpcSubnet{ - Tags: newTags, - } - } else { - updatedSubnet = &model.VpcSubnet{ - Tags: newTags, - SubnetDhcpConfig: service.buildSubnetDHCPConfig(dhcpMode), - } + // Avoid updating vpcSubnets[i] to ensure Subnet store + // is only updated after the updating succeeds. + updatedSubnet := *vpcSubnets[i] + updatedSubnet.Tags = newTags + // Update the SubnetSet DHCP Config + if updatedSubnet.SubnetDhcpConfig != nil { + // Generate a new SubnetDhcpConfig for updatedSubnet to + // avoid changing vpcSubnets[i].SubnetDhcpConfig + updatedSubnet.SubnetDhcpConfig = service.buildSubnetDHCPConfig(dhcpMode, updatedSubnet.SubnetDhcpConfig.DhcpServerAdditionalConfig) } - changed := common.CompareResource(SubnetToComparable(vpcSubnets[i]), SubnetToComparable(updatedSubnet)) + changed := common.CompareResource(SubnetToComparable(vpcSubnets[i]), SubnetToComparable(&updatedSubnet)) if !changed { log.Info("NSX Subnet unchanged, skipping update", "Subnet", *vpcSubnet.Id) continue } - vpcSubnets[i].Tags = newTags - // Update the SubnetSet DHCP Config - if vpcSubnets[i].SubnetDhcpConfig != nil { - vpcSubnets[i].SubnetDhcpConfig = service.buildSubnetDHCPConfig(dhcpMode) - vpcSubnets[i].AdvancedConfig.StaticIpAllocation.Enabled = &staticIpAllocation - } - vpcInfo, err := common.ParseVPCResourcePath(*vpcSubnets[i].Path) if err != nil { err := fmt.Errorf("failed to parse NSX VPC path for Subnet %s: %s", *vpcSubnets[i].Path, err) return err } - if _, err := service.createOrUpdateSubnet(subnetSet, vpcSubnets[i], &vpcInfo); err != nil { + if _, err := service.createOrUpdateSubnet(subnetSet, &updatedSubnet, &vpcInfo); err != nil { return fmt.Errorf("failed to update Subnet %s in SubnetSet %s: %w", *vpcSubnet.Id, subnetSet.Name, err) } log.Info("Successfully updated SubnetSet", "subnetSet", subnetSet, "Subnet", *vpcSubnet.Id) diff --git a/pkg/nsx/services/subnet/subnet_test.go b/pkg/nsx/services/subnet/subnet_test.go index 9fc11ce8b..3ad8186dd 100644 --- a/pkg/nsx/services/subnet/subnet_test.go +++ b/pkg/nsx/services/subnet/subnet_test.go @@ -2,6 +2,7 @@ package subnet import ( "context" + "errors" "reflect" "testing" @@ -15,6 +16,7 @@ import ( "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apimachinery/pkg/util/wait" clientgoscheme "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/tools/cache" "sigs.k8s.io/controller-runtime/pkg/client" @@ -25,6 +27,7 @@ import ( mock_client "github.com/vmware-tanzu/nsx-operator/pkg/mock/controller-runtime/client" "github.com/vmware-tanzu/nsx-operator/pkg/nsx" "github.com/vmware-tanzu/nsx-operator/pkg/nsx/services/common" + "github.com/vmware-tanzu/nsx-operator/pkg/nsx/services/realizestate" "github.com/vmware-tanzu/nsx-operator/pkg/util" ) @@ -135,6 +138,13 @@ func (f fakeSubnetsClient) Update(orgIdParam string, projectIdParam string, vpcI return model.VpcSubnet{}, nil } +type fakeSubnetStatusClient struct { +} + +func (f fakeSubnetStatusClient) List(orgIdParam string, projectIdParam string, vpcIdParam string, subnetIdParam string) (model.VpcSubnetStatusListResult, error) { + return model.VpcSubnetStatusListResult{}, nil +} + type fakeRealizedEntitiesClient struct { } @@ -151,6 +161,21 @@ func (f fakeRealizedEntitiesClient) List(intentPathParam string, sitePathParam * }, nil } +type fakeStatusWriter struct { +} + +func (writer fakeStatusWriter) Create(ctx context.Context, obj client.Object, subResource client.Object, opts ...client.SubResourceCreateOption) error { + return nil +} + +func (writer fakeStatusWriter) Update(ctx context.Context, obj client.Object, opts ...client.SubResourceUpdateOption) error { + return nil +} + +func (writer fakeStatusWriter) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + return nil +} + func TestInitializeSubnetService(t *testing.T) { clusterName := "k8scl-one:test" subnetID := "fakeSubnetUID" @@ -406,3 +431,127 @@ func TestSubnetService_UpdateSubnetSet(t *testing.T) { err := service.UpdateSubnetSet("ns-1", vpcSubnets, tags, "") assert.Nil(t, err) } + +func TestSubnetService_createOrUpdateSubnet(t *testing.T) { + mockCtl := gomock.NewController(t) + k8sClient := mock_client.NewMockClient(mockCtl) + defer mockCtl.Finish() + service := &SubnetService{ + Service: common.Service{ + Client: k8sClient, + NSXClient: &nsx.Client{ + OrgRootClient: &fakeOrgRootClient{}, + SubnetsClient: &fakeSubnetsClient{}, + SubnetStatusClient: &fakeSubnetStatusClient{}, + }, + }, + SubnetStore: &SubnetStore{ + ResourceStore: common.ResourceStore{ + Indexer: cache.NewIndexer(keyFunc, cache.Indexers{ + common.TagScopeSubnetCRUID: subnetIndexFunc, + common.TagScopeSubnetSetCRUID: subnetSetIndexFunc, + common.TagScopeVMNamespace: subnetIndexVMNamespaceFunc, + common.TagScopeNamespace: subnetIndexNamespaceFunc, + }), + BindingType: model.VpcSubnetBindingType(), + }, + }, + } + + fakeSubnet := model.VpcSubnet{ + Id: common.String("subnet-1"), + Path: common.String("/orgs/default/projects/default/vpcs/default/subnets/subnet-path-1"), + Tags: []model.Tag{ + { + Scope: common.String(common.TagScopeSubnetSetCRUID), + Tag: common.String("subnetset-1"), + }, + }, + } + fakewriter := &fakeStatusWriter{} + + testCases := []struct { + name string + prepareFunc func() *gomonkey.Patches + expectedErr string + crObj client.Object + }{ + { + name: "Update Subnet with RealizedState and deletion error", + prepareFunc: func() *gomonkey.Patches { + patches := gomonkey.ApplyFunc((*realizestate.RealizeStateService).CheckRealizeState, + func(_ *realizestate.RealizeStateService, _ wait.Backoff, _ string) error { + return realizestate.NewRealizeStateError("mocked realized error") + }) + patches.ApplyFunc((*SubnetService).DeleteSubnet, func(_ *SubnetService, _ model.VpcSubnet) error { + return errors.New("mocked deletion error") + }) + patches.ApplyFunc(fakeSubnetsClient.Get, func(f fakeSubnetsClient, orgIdParam string, projectIdParam string, vpcIdParam string, subnetIdParam string) (model.VpcSubnet, error) { + return fakeSubnet, nil + }) + return patches + }, + crObj: &v1alpha1.Subnet{}, + expectedErr: "realization check failed: mocked realized error; deletion failed: mocked deletion error", + }, + { + name: "Create Subnet for SubnetSet Success", + prepareFunc: func() *gomonkey.Patches { + patches := gomonkey.ApplyFunc((*realizestate.RealizeStateService).CheckRealizeState, + func(_ *realizestate.RealizeStateService, _ wait.Backoff, _ string) error { + return nil + }) + patches.ApplyFunc(fakeSubnetsClient.Get, + func(f fakeSubnetsClient, orgIdParam string, projectIdParam string, vpcIdParam string, subnetIdParam string) (model.VpcSubnet, error) { + return fakeSubnet, nil + }) + patches.ApplyFunc(fakeSubnetStatusClient.List, + func(_ fakeSubnetStatusClient, orgIdParam string, projectIdParam string, vpcIdParam string, subnetIdParam string) (model.VpcSubnetStatusListResult, error) { + return model.VpcSubnetStatusListResult{ + Results: []model.VpcSubnetStatus{ + { + NetworkAddress: common.String("10.0.0.0/28"), + GatewayAddress: common.String("10.0.0.1/28"), + DhcpServerAddress: common.String("10.0.0.2/28"), + }, + }, + }, nil + }) + k8sClient.EXPECT().Status().Return(fakewriter) + patches.ApplyFunc(fakeStatusWriter.Update, + func(writer fakeStatusWriter, ctx context.Context, obj client.Object, opts ...client.SubResourceUpdateOption) error { + subnetSet := obj.(*v1alpha1.SubnetSet) + assert.Equal(t, 1, len(subnetSet.Status.Subnets)) + assert.Equal(t, "10.0.0.0/28", subnetSet.Status.Subnets[0].NetworkAddresses[0]) + assert.Equal(t, "10.0.0.1/28", subnetSet.Status.Subnets[0].GatewayAddresses[0]) + assert.Equal(t, "10.0.0.2/28", subnetSet.Status.Subnets[0].DHCPServerAddresses[0]) + return nil + }) + return patches + }, + crObj: &v1alpha1.SubnetSet{ + ObjectMeta: metav1.ObjectMeta{UID: "subnetset-1"}, + }, + }, + } + + for _, tt := range testCases { + t.Run(tt.name, func(t *testing.T) { + if tt.prepareFunc != nil { + patches := tt.prepareFunc() + defer patches.Reset() + } + nsxSubnet, err := service.createOrUpdateSubnet( + tt.crObj, + &fakeSubnet, + &common.VPCResourceInfo{}, + ) + if tt.expectedErr != "" { + assert.Equal(t, tt.expectedErr, err.Error()) + } else { + assert.Nil(t, err) + assert.Equal(t, fakeSubnet, *nsxSubnet) + } + }) + } +}