Fixed target group leaking problems (#60)

* Fixed the target group leaking issue

* Fixed target_group_manager unit test error

* Fixed AWS resource tag to determine if a target group is stale

* Add unit test

* Update tg synthesizer unit test to align with current logic

* Update the logging

* Add more unit tests to cover serviceexport

* Make sure retry deleting stale target group when its targets are in
draining state

* make sure stale rules get deleted during deleting stale target groups

* Reconcile every 10 min

* Address code review comments

Author: liwenwu-amazon

pkg/deploy/lattice/target_group_manager.go

@@ -16,7 +16,7 @@ import (
 type TargetGroupManager interface {
 	Create(ctx context.Context, targetGroup *latticemodel.TargetGroup) (latticemodel.TargetGroupStatus, error)
 	Delete(ctx context.Context, targetGroup *latticemodel.TargetGroup) error
-	List(ctx context.Context) ([]vpclattice.GetTargetGroupOutput, error)
+	List(ctx context.Context) ([]targetGroupOutput, error)
 	Get(tx context.Context, targetGroup *latticemodel.TargetGroup) (latticemodel.TargetGroupStatus, error)
 }
 
@@ -189,9 +189,14 @@ func (s *defaultTargetGroupManager) Delete(ctx context.Context, targetGroup *lat
 	return err
 }
 
-func (s *defaultTargetGroupManager) List(ctx context.Context) ([]vpclattice.GetTargetGroupOutput, error) {
+type targetGroupOutput struct {
+	getTargetGroupOutput vpclattice.GetTargetGroupOutput
+	targetGroupTags      *vpclattice.ListTagsForResourceOutput
+}
+
+func (s *defaultTargetGroupManager) List(ctx context.Context) ([]targetGroupOutput, error) {
 	vpcLatticeSess := s.cloud.Lattice()
-	var tgList []vpclattice.GetTargetGroupOutput
+	var tgList []targetGroupOutput
 	targetGroupListInput := vpclattice.ListTargetGroupsInput{}
 	resp, err := vpcLatticeSess.ListTargetGroupsAsList(ctx, &targetGroupListInput)
 	glog.V(6).Infof("ManagerList: %v, err: %v \n", resp, err)
@@ -213,7 +218,25 @@ func (s *defaultTargetGroupManager) List(ctx context.Context) ([]vpclattice.GetT
 		//glog.V(6).Infof("Manager-List: tg-vpc %v , config.vpc %v\n", aws.StringValue(tgOutput.Config.VpcId), config.VpcID)
 
 		if tgOutput.Config != nil && aws.StringValue(tgOutput.Config.VpcIdentifier) == config.VpcID {
-			tgList = append(tgList, *tgOutput)
+			// retrieve target group tags
+			//ListTagsForResourceWithContext
+			tagsInput := vpclattice.ListTagsForResourceInput{
+				ResourceArn: tg.Arn,
+			}
+
+			tagsOutput, err := vpcLatticeSess.ListTagsForResourceWithContext(ctx, &tagsInput)
+
+			glog.V(6).Infof("tagsOutput %v,  err: %v", tagsOutput, err)
+
+			if err != nil {
+				// setting it to nil, so the caller knows there is tag resource associated to this target group
+				tagsOutput = nil
+			}
+			tgOutput := targetGroupOutput{
+				getTargetGroupOutput: *tgOutput,
+				targetGroupTags:      tagsOutput,
+			}
+			tgList = append(tgList, tgOutput)
 		}
 	}
 	return tgList, err

pkg/deploy/lattice/target_group_manager_mock.go

@@ -813,12 +813,20 @@ func Test_ListTG_TGsExist(t *testing.T) {
 	mockCloud := mocks_aws.NewMockCloud(c)
 	mockVpcLatticeSess.EXPECT().ListTargetGroupsAsList(ctx, gomock.Any()).Return(listTGOutput, nil)
 	mockVpcLatticeSess.EXPECT().GetTargetGroupWithContext(ctx, gomock.Any()).Return(getTG1, nil)
+	// assume no tags
+	mockVpcLatticeSess.EXPECT().ListTagsForResourceWithContext(ctx, gomock.Any()).Return(nil, errors.New("no tags"))
 	mockVpcLatticeSess.EXPECT().GetTargetGroupWithContext(ctx, gomock.Any()).Return(getTG2, nil)
+
 	mockCloud.EXPECT().Lattice().Return(mockVpcLatticeSess)
 
 	tgManager := NewTargetGroupManager(mockCloud)
 	tgList, err := tgManager.List(ctx)
-	expect := []vpclattice.GetTargetGroupOutput{*getTG1}
+	expect := []targetGroupOutput{
+		{
+			getTargetGroupOutput: *getTG1,
+			targetGroupTags:      nil,
+		},
+	}
 
 	assert.Nil(t, err)
 	assert.Equal(t, tgList, expect)
@@ -837,9 +845,10 @@ func Test_ListTG_NoTG(t *testing.T) {
 
 	tgManager := NewTargetGroupManager(mockCloud)
 	tgList, err := tgManager.List(ctx)
+	expectTgList := []targetGroupOutput(nil)
 
 	assert.Nil(t, err)
-	assert.Equal(t, tgList, []vpclattice.GetTargetGroupOutput(nil))
+	assert.Equal(t, tgList, expectTgList)
 }
 
 func Test_Get(t *testing.T) {

pkg/deploy/lattice/target_group_manager_test.go

@@ -6,8 +6,10 @@ import (
 	"github.com/golang/glog"
 	"strings"
 
+	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/gateway-api/apis/v1alpha2"
+	mcs_api "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 
 	lattice_aws "github.com/aws/aws-application-networking-k8s/pkg/aws"
 	"github.com/aws/aws-application-networking-k8s/pkg/config"
@@ -45,10 +47,10 @@ func (t *targetGroupSynthesizer) Synthesize(ctx context.Context) error {
 
 	/* TODO,  resolve bug that this might delete other HTTPRoute's TG before they have chance
 	 * to be reconcile during controller restart
+	 */
 	if err := t.SynthesizeSDKTargetGroups(ctx); err != nil {
 		ret = LATTICE_RETRY
 	}
-	*/
 
 	if ret != "" {
 		return errors.New(ret)
@@ -158,48 +160,140 @@ func (t *targetGroupSynthesizer) SynthesizeSDKTargetGroups(ctx context.Context)
 	sdkTGs, err := t.targetGroupManager.List(ctx)
 
 	if err != nil {
-		glog.V(6).Infof("SynthesizeSDKTargetGroups: failed to retrieve sdk TGs %v\n", err)
+		glog.V(2).Infof("SynthesizeSDKTargetGroups: failed to retrieve sdk TGs %v\n", err)
 		return nil
 	}
 
 	glog.V(6).Infof("SynthesizeSDKTargetGroups: here is sdkTGs %v len %v \n", sdkTGs, len(sdkTGs))
 
 	for _, sdkTG := range sdkTGs {
 
-		if *sdkTG.Config.VpcIdentifier != config.VpcID {
-			glog.V(6).Infof("Ignore target group for other VPCs, other :%v, current vpc %v\n", *sdkTG.Config.VpcIdentifier, config.VpcID)
+		if *sdkTG.getTargetGroupOutput.Config.VpcIdentifier != config.VpcID {
+			glog.V(6).Infof("Ignore target group ARN %v Name %v for other VPCs",
+				*sdkTG.getTargetGroupOutput.Arn, *sdkTG.getTargetGroupOutput.Name)
 			continue
 		}
-		if tg, err := t.latticeDataStore.GetTargetGroup(*sdkTG.Name, true); err == nil {
-			glog.V(6).Infof("Ignore target group created by service import %v\n", tg)
+
+		// does target group have K8S tags,  ignore if it is not tagged
+		tgTags := sdkTG.targetGroupTags
+
+		if tgTags == nil || tgTags.Tags == nil {
+			glog.V(6).Infof("Ignore target group not tagged for K8S, %v, %v \n",
+				*sdkTG.getTargetGroupOutput.Arn, *sdkTG.getTargetGroupOutput.Name)
 			continue
 		}
 
-		tg, err := t.latticeDataStore.GetTargetGroup(*sdkTG.Name, false)
-		if err != nil {
-			staleSDKTGs = append(staleSDKTGs, latticemodel.TargetGroup{
-				Spec: latticemodel.TargetGroupSpec{
-					Name:      *sdkTG.Name,
-					LatticeID: *sdkTG.Id,
-				},
-			})
-		} else {
-			if tg.ByServiceExport {
+		parentRef, ok := tgTags.Tags[latticemodel.K8SParentRefTypeKey]
+		if !ok || parentRef == nil {
+			glog.V(6).Infof("Ignore target group that have no K8S parentRef tag :%v, %v \n",
+				*sdkTG.getTargetGroupOutput.Arn, *sdkTG.getTargetGroupOutput.Name)
+			continue
+		}
+
+		srvName, ok := tgTags.Tags[latticemodel.K8SServiceNameKey]
+
+		if !ok || srvName == nil {
+			glog.V(6).Infof("Ignore TargetGroup have no servicename tag: %v, %v",
+				*sdkTG.getTargetGroupOutput.Arn, *sdkTG.getTargetGroupOutput.Name)
+			continue
+		}
+
+		srvNamespace, ok := tgTags.Tags[latticemodel.K8SServiceNamespaceKey]
+
+		if !ok || srvNamespace == nil {
+			glog.V(6).Infof("Ignore TargetGroup have no servicenamespace tag: %v, %v",
+				*sdkTG.getTargetGroupOutput.Arn, *sdkTG.getTargetGroupOutput.Name)
+			continue
+		}
+
+		// if its parentref is service export,  check the parent service export exist
+		// Ignore if service export exists
+		if *parentRef == latticemodel.K8SServiceExportType {
+			glog.V(6).Infof("TargetGroup %v, %v is referenced by ServiceExport",
+				*sdkTG.getTargetGroupOutput.Arn, *sdkTG.getTargetGroupOutput.Name)
+
+			glog.V(6).Infof("Determine serviceexport name=%v, namespace=%v exists for targetGroup %v",
+				*srvName, *srvNamespace, *sdkTG.getTargetGroupOutput.Arn)
+
+			srvExportName := types.NamespacedName{
+				Namespace: *srvNamespace,
+				Name:      *srvName,
+			}
+			srvExport := &mcs_api.ServiceExport{}
+			if err := t.client.Get(ctx, srvExportName, srvExport); err == nil {
+
+				glog.V(6).Infof("Ignore TargetGroup(triggered by serviceexport) %v, %v since serviceexport object is found",
+					*sdkTG.getTargetGroupOutput.Arn, *sdkTG.getTargetGroupOutput.Name)
 				continue
 			}
+		}
+
+		// if its parentref is HTTP/route, check the parent HTTPRoute exist
+		// Ignore if httpRoute does NOT exist
+		if *parentRef == latticemodel.K8SHTTPRouteType {
+			glog.V(6).Infof("TargetGroup %v, %v is referenced by HTTPRoute",
+				*sdkTG.getTargetGroupOutput.Arn, *sdkTG.getTargetGroupOutput.Name)
 
-			if t.isTargetGroupUsedByHTTPRoute(ctx, *sdkTG.Name) {
+			httpName, ok := tgTags.Tags[latticemodel.K8SHTTPRouteNameKey]
+
+			if !ok || httpName == nil {
+				glog.V(6).Infof("Ignore TargetGroup(triggered by httpRoute) %v, %v have no httproute name tag",
+					*sdkTG.getTargetGroupOutput.Arn, *sdkTG.getTargetGroupOutput.Name)
 				continue
 			}
-			staleSDKTGs = append(staleSDKTGs, latticemodel.TargetGroup{
-				Spec: latticemodel.TargetGroupSpec{
-					Name:      *sdkTG.Name,
-					LatticeID: *sdkTG.Id,
-				},
-			})
 
+			httpNamespace, ok := tgTags.Tags[latticemodel.K8SHTTPRouteNamespaceKey]
+
+			if !ok || httpNamespace == nil {
+				glog.V(6).Infof("Ignore TargetGroup(triggered by httpRoute) %v, %v have no httproute namespace tag",
+					*sdkTG.getTargetGroupOutput.Arn, *sdkTG.getTargetGroupOutput.Name)
+				continue
+			}
+
+			httprouteName := types.NamespacedName{
+				Namespace: *httpNamespace,
+				Name:      *httpName,
+			}
+
+			httpRoute := &v1alpha2.HTTPRoute{}
+
+			tgName := latticestore.TargetGroupName(*srvName, *srvNamespace)
+
+			if err := t.client.Get(ctx, httprouteName, httpRoute); err != nil {
+				glog.V(6).Infof("tgname %v is not used by httproute %v\n", tgName, httpRoute)
+
+			} else {
+
+				isUsed := t.isTargetGroupUsedByaHTTPRoute(ctx, tgName, httpRoute)
+
+				if isUsed {
+
+					glog.V(6).Infof("Ignore TargetGroup(triggered by HTTProute) %v, %v since httproute object is found",
+						*sdkTG.getTargetGroupOutput.Arn, *sdkTG.getTargetGroupOutput.Name)
+
+					continue
+				} else {
+					glog.V(6).Infof("tgname %v is not used by httproute %v\n", tgName, httpRoute)
+				}
+			}
+
+		}
+
+		if tg, err := t.latticeDataStore.GetTargetGroup(*sdkTG.getTargetGroupOutput.Name, true); err == nil {
+			glog.V(6).Infof("Ignore target group created by service import %v\n", tg)
+			continue
 		}
 
+		glog.V(2).Infof("Append stale SDK TG to stale list Name %v, ARN %v",
+			*sdkTG.getTargetGroupOutput.Name, *sdkTG.getTargetGroupOutput.Id)
+
+		staleSDKTGs = append(staleSDKTGs, latticemodel.TargetGroup{
+			Spec: latticemodel.TargetGroupSpec{
+				Name:      *sdkTG.getTargetGroupOutput.Name,
+				LatticeID: *sdkTG.getTargetGroupOutput.Id,
+			},
+		})
+
 	}
 
 	glog.V(6).Infof("SynthesizeSDKTargetGroups, here is the stale target groups list %v stalelen %d\n", staleSDKTGs, len(staleSDKTGs))
@@ -211,8 +305,7 @@ func (t *targetGroupSynthesizer) SynthesizeSDKTargetGroups(ctx context.Context)
 		err := t.targetGroupManager.Delete(ctx, &sdkTG)
 		glog.V(6).Infof("SynthesizeSDKTargetGroups, deleting stale target group %v \n", err)
 
-		// TODO find out the error code
-		if err != nil && !strings.Contains(err.Error(), "ConflictException") {
+		if err != nil && !strings.Contains(err.Error(), "TargetGroup is referenced in routing configuration, listeners or rules of service.") {
 			ret_err = true
 		}
 		// continue on even when there is an err
@@ -227,35 +320,24 @@ func (t *targetGroupSynthesizer) SynthesizeSDKTargetGroups(ctx context.Context)
 
 }
 
-// TODO put following routine to common library
-func (t *targetGroupSynthesizer) isTargetGroupUsedByHTTPRoute(ctx context.Context, tgName string) bool {
-
-	httpRouteList := &v1alpha2.HTTPRouteList{}
-
-	t.client.List(ctx, httpRouteList)
-
-	//glog.V(6).Infof("isTargetGroupUsedByHTTPRoute: tgName %v-- %v\n", tgName, httpRouteList)
-
-	for _, httpRoute := range httpRouteList.Items {
-		for _, httpRule := range httpRoute.Spec.Rules {
-			for _, httpBackendRef := range httpRule.BackendRefs {
-				//glog.V(6).Infof("isTargetGroupUsedByHTTPRoute: httpBackendRef: %v \n", httpBackendRef)
-				if string(*httpBackendRef.BackendObjectReference.Kind) != "Service" {
-					continue
-				}
-				namespace := "default"
+func (t *targetGroupSynthesizer) isTargetGroupUsedByaHTTPRoute(ctx context.Context, tgName string, httpRoute *v1alpha2.HTTPRoute) bool {
 
-				if httpBackendRef.BackendObjectReference.Namespace != nil {
-					namespace = string(*httpBackendRef.BackendObjectReference.Namespace)
-				}
-				refTGName := latticestore.TargetGroupName(string(httpBackendRef.BackendObjectReference.Name), namespace)
-				//glog.V(6).Infof("refTGName: %s\n", refTGName)
+	for _, httpRule := range httpRoute.Spec.Rules {
+		for _, httpBackendRef := range httpRule.BackendRefs {
+			if string(*httpBackendRef.BackendObjectReference.Kind) != "Service" {
+				continue
+			}
+			namespace := "default"
 
-				if tgName == refTGName {
-					return true
-				}
+			if httpBackendRef.BackendObjectReference.Namespace != nil {
+				namespace = string(*httpBackendRef.BackendObjectReference.Namespace)
+			}
+			refTGName := latticestore.TargetGroupName(string(httpBackendRef.BackendObjectReference.Name), namespace)
 
+			if tgName == refTGName {
+				return true
 			}
+
 		}
 	}