Merge pull request #2510 from voro015/aws-sd-delete-empty-services

AWSSD: Cleanup empty Services

main.go

@@ -206,7 +206,7 @@ func main() {
 			log.Infof("Registry \"%s\" cannot be used with AWS Cloud Map. Switching to \"aws-sd\".", cfg.Registry)
 			cfg.Registry = "aws-sd"
 		}
-		p, err = awssd.NewAWSSDProvider(domainFilter, cfg.AWSZoneType, cfg.AWSAssumeRole, cfg.DryRun)
+		p, err = awssd.NewAWSSDProvider(domainFilter, cfg.AWSZoneType, cfg.AWSAssumeRole, cfg.DryRun, cfg.AWSSDServiceCleanup, cfg.TXTOwnerID)
 	case "azure-dns", "azure":
 		p, err = azure.NewAzureProvider(cfg.AzureConfigFile, domainFilter, zoneNameFilter, zoneIDFilter, cfg.AzureResourceGroup, cfg.AzureUserAssignedIdentityClientID, cfg.DryRun)
 	case "azure-private-dns":

pkg/apis/externaldns/types.go

@@ -87,6 +87,7 @@ type Config struct {
 	AWSAPIRetries                     int
 	AWSPreferCNAME                    bool
 	AWSZoneCacheDuration              time.Duration
+	AWSSDServiceCleanup               bool
 	AzureConfigFile                   string
 	AzureResourceGroup                string
 	AzureSubscriptionID               string
@@ -226,6 +227,7 @@ var defaultConfig = &Config{
 	AWSAPIRetries:               3,
 	AWSPreferCNAME:              false,
 	AWSZoneCacheDuration:        0 * time.Second,
+	AWSSDServiceCleanup:         false,
 	AzureConfigFile:             "/etc/kubernetes/azure.json",
 	AzureResourceGroup:          "",
 	AzureSubscriptionID:         "",
@@ -417,6 +419,7 @@ func (cfg *Config) ParseFlags(args []string) error {
 	app.Flag("aws-api-retries", "When using the AWS provider, set the maximum number of retries for API calls before giving up.").Default(strconv.Itoa(defaultConfig.AWSAPIRetries)).IntVar(&cfg.AWSAPIRetries)
 	app.Flag("aws-prefer-cname", "When using the AWS provider, prefer using CNAME instead of ALIAS (default: disabled)").BoolVar(&cfg.AWSPreferCNAME)
 	app.Flag("aws-zones-cache-duration", "When using the AWS provider, set the zones list cache TTL (0s to disable).").Default(defaultConfig.AWSZoneCacheDuration.String()).DurationVar(&cfg.AWSZoneCacheDuration)
+	app.Flag("aws-sd-service-cleanup", "When using the AWS CloudMap provider, delete empty Services without endpoints (default: disabled)").BoolVar(&cfg.AWSSDServiceCleanup)
 	app.Flag("azure-config-file", "When using the Azure provider, specify the Azure configuration file (required when --provider=azure").Default(defaultConfig.AzureConfigFile).StringVar(&cfg.AzureConfigFile)
 	app.Flag("azure-resource-group", "When using the Azure provider, override the Azure resource group to use (required when --provider=azure-private-dns)").Default(defaultConfig.AzureResourceGroup).StringVar(&cfg.AzureResourceGroup)
 	app.Flag("azure-subscription-id", "When using the Azure provider, specify the Azure configuration file (required when --provider=azure-private-dns)").Default(defaultConfig.AzureSubscriptionID).StringVar(&cfg.AzureSubscriptionID)

pkg/apis/externaldns/types_test.go

@@ -63,6 +63,7 @@ var (
 		AWSAPIRetries:               3,
 		AWSPreferCNAME:              false,
 		AWSZoneCacheDuration:        0 * time.Second,
+		AWSSDServiceCleanup:         false,
 		AzureConfigFile:             "/etc/kubernetes/azure.json",
 		AzureResourceGroup:          "",
 		AzureSubscriptionID:         "",
@@ -160,6 +161,7 @@ var (
 		AWSAPIRetries:               13,
 		AWSPreferCNAME:              true,
 		AWSZoneCacheDuration:        10 * time.Second,
+		AWSSDServiceCleanup:         true,
 		AzureConfigFile:             "azure.json",
 		AzureResourceGroup:          "arg",
 		AzureSubscriptionID:         "arg",
@@ -325,6 +327,7 @@ func TestParseFlags(t *testing.T) {
 				"--aws-api-retries=13",
 				"--aws-prefer-cname",
 				"--aws-zones-cache-duration=10s",
+				"--aws-sd-service-cleanup",
 				"--no-aws-evaluate-target-health",
 				"--policy=upsert-only",
 				"--registry=noop",
@@ -435,6 +438,7 @@ func TestParseFlags(t *testing.T) {
 				"EXTERNAL_DNS_AWS_API_RETRIES":                 "13",
 				"EXTERNAL_DNS_AWS_PREFER_CNAME":                "true",
 				"EXTERNAL_DNS_AWS_ZONES_CACHE_DURATION":        "10s",
+				"EXTERNAL_DNS_AWS_SD_SERVICE_CLEANUP":          "true",
 				"EXTERNAL_DNS_POLICY":                          "upsert-only",
 				"EXTERNAL_DNS_REGISTRY":                        "noop",
 				"EXTERNAL_DNS_TXT_OWNER_ID":                    "owner-1",

provider/awssd/aws_sd.go

@@ -69,6 +69,7 @@ type AWSSDClient interface {
 	ListServicesPages(input *sd.ListServicesInput, fn func(*sd.ListServicesOutput, bool) bool) error
 	RegisterInstance(input *sd.RegisterInstanceInput) (*sd.RegisterInstanceOutput, error)
 	UpdateService(input *sd.UpdateServiceInput) (*sd.UpdateServiceOutput, error)
+	DeleteService(input *sd.DeleteServiceInput) (*sd.DeleteServiceOutput, error)
 }
 
 // AWSSDProvider is an implementation of Provider for AWS Cloud Map.
@@ -80,10 +81,14 @@ type AWSSDProvider struct {
 	namespaceFilter endpoint.DomainFilter
 	// filter namespace by type (private or public)
 	namespaceTypeFilter *sd.NamespaceFilter
+	// enables service without instances cleanup
+	cleanEmptyService bool
+	// filter services for removal
+	ownerID string
 }
 
 // NewAWSSDProvider initializes a new AWS Cloud Map based Provider.
-func NewAWSSDProvider(domainFilter endpoint.DomainFilter, namespaceType string, assumeRole string, dryRun bool) (*AWSSDProvider, error) {
+func NewAWSSDProvider(domainFilter endpoint.DomainFilter, namespaceType string, assumeRole string, dryRun, cleanEmptyService bool, ownerID string) (*AWSSDProvider, error) {
 	config := aws.NewConfig()
 
 	config = config.WithHTTPClient(
@@ -112,9 +117,11 @@ func NewAWSSDProvider(domainFilter endpoint.DomainFilter, namespaceType string,
 
 	provider := &AWSSDProvider{
 		client:              sd.New(sess),
+		dryRun:              dryRun,
 		namespaceFilter:     domainFilter,
 		namespaceTypeFilter: newSdNamespaceFilter(namespaceType),
-		dryRun:              dryRun,
+		cleanEmptyService:   cleanEmptyService,
+		ownerID:             ownerID,
 	}
 
 	return provider, nil
@@ -161,6 +168,12 @@ func (p *AWSSDProvider) Records(ctx context.Context) (endpoints []*endpoint.Endp
 				ep := p.instancesToEndpoint(ns, srv, instances)
 				endpoints = append(endpoints, ep)
 			}
+			if len(instances) == 0 {
+				err = p.DeleteService(srv)
+				if err != nil {
+					log.Warnf("Failed to delete service \"%s\", error: %s", aws.StringValue(srv.Name), err)
+				}
+			}
 		}
 	}
 
@@ -284,9 +297,8 @@ func (p *AWSSDProvider) submitCreates(namespaces []*sd.NamespaceSummary, changes
 				}
 				// update local list of services
 				services[*srv.Name] = srv
-			} else if (ch.RecordTTL.IsConfigured() && *srv.DnsConfig.DnsRecords[0].TTL != int64(ch.RecordTTL)) ||
+			} else if ch.RecordTTL.IsConfigured() && *srv.DnsConfig.DnsRecords[0].TTL != int64(ch.RecordTTL) {
-				aws.StringValue(srv.Description) != ch.Labels[endpoint.AWSSDDescriptionLabel] {
+				// update service when TTL differ
-				// update service when TTL or Description differ
 				err = p.UpdateService(srv, ch)
 				if err != nil {
 					return err
@@ -483,6 +495,27 @@ func (p *AWSSDProvider) UpdateService(service *sd.Service, ep *endpoint.Endpoint
 	return nil
 }
 
+// DeleteService deletes empty Service from AWS API if its owner id match
+func (p *AWSSDProvider) DeleteService(service *sd.Service) error {
+	log.Debugf("Check if service \"%s\" owner id match and it can be deleted", *service.Name)
+	if !p.dryRun && p.cleanEmptyService {
+		// convert ownerID string to service description format
+		label := endpoint.NewLabels()
+		label[endpoint.OwnerLabelKey] = p.ownerID
+		label[endpoint.AWSSDDescriptionLabel] = label.Serialize(false)
+
+		if aws.StringValue(service.Description) == label[endpoint.AWSSDDescriptionLabel] {
+			log.Infof("Deleting service \"%s\"", *service.Name)
+			_, err := p.client.DeleteService(&sd.DeleteServiceInput{
+				Id: aws.String(*service.Id),
+			})
+			return err
+		}
+		log.Debugf("Skipping service removal %s because owner id does not match, found: \"%s\", required: \"%s\"", aws.StringValue(service.Name), aws.StringValue(service.Description), label[endpoint.AWSSDDescriptionLabel])
+	}
+	return nil
+}
+
 // RegisterInstance creates a new instance in given service.
 func (p *AWSSDProvider) RegisterInstance(service *sd.Service, ep *endpoint.Endpoint) error {
 	for _, target := range ep.Targets {

provider/awssd/aws_sd_test.go

@@ -181,12 +181,27 @@ func (s *AWSSDClientStub) UpdateService(input *sd.UpdateServiceInput) (*sd.Updat
 	return &sd.UpdateServiceOutput{}, nil
 }
 
-func newTestAWSSDProvider(api AWSSDClient, domainFilter endpoint.DomainFilter, namespaceTypeFilter string) *AWSSDProvider {
+func (s *AWSSDClientStub) DeleteService(input *sd.DeleteServiceInput) (*sd.DeleteServiceOutput, error) {
+	out, err := s.GetService(&sd.GetServiceInput{Id: input.Id})
+	if err != nil {
+		return nil, err
+	}
+
+	service := out.Service
+	namespace := s.services[*service.NamespaceId]
+	delete(namespace, *input.Id)
+
+	return &sd.DeleteServiceOutput{}, nil
+}
+
+func newTestAWSSDProvider(api AWSSDClient, domainFilter endpoint.DomainFilter, namespaceTypeFilter, ownerID string) *AWSSDProvider {
 	return &AWSSDProvider{
 		client:              api,
+		dryRun:              false,
 		namespaceFilter:     domainFilter,
 		namespaceTypeFilter: newSdNamespaceFilter(namespaceTypeFilter),
-		dryRun:              false,
+		cleanEmptyService:   true,
+		ownerID:             ownerID,
 	}
 }
 
@@ -288,7 +303,7 @@ func TestAWSSDProvider_Records(t *testing.T) {
 		instances:  instances,
 	}
 
-	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "")
+	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
 
 	endpoints, _ := provider.Records(context.Background())
 
@@ -316,7 +331,7 @@ func TestAWSSDProvider_ApplyChanges(t *testing.T) {
 		{DNSName: "service3.private.com", Targets: endpoint.Targets{"cname.target.com"}, RecordType: endpoint.RecordTypeCNAME, RecordTTL: 100},
 	}
 
-	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "")
+	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
 
 	ctx := context.Background()
 
@@ -376,7 +391,7 @@ func TestAWSSDProvider_ListNamespaces(t *testing.T) {
 		{"domain filter", endpoint.NewDomainFilter([]string{"public.com"}), "", []*sd.NamespaceSummary{namespaceToNamespaceSummary(namespaces["public"])}},
 		{"non-existing domain", endpoint.NewDomainFilter([]string{"xxx.com"}), "", []*sd.NamespaceSummary{}},
 	} {
-		provider := newTestAWSSDProvider(api, tc.domainFilter, tc.namespaceTypeFilter)
+		provider := newTestAWSSDProvider(api, tc.domainFilter, tc.namespaceTypeFilter, "")
 
 		result, err := provider.ListNamespaces()
 		require.NoError(t, err)
@@ -442,7 +457,7 @@ func TestAWSSDProvider_ListServicesByNamespace(t *testing.T) {
 	}{
 		{map[string]*sd.Service{"service1": services["private"]["srv1"], "service2": services["private"]["srv2"]}},
 	} {
-		provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "")
+		provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
 
 		result, err := provider.ListServicesByNamespaceID(namespaces["private"].Id)
 		require.NoError(t, err)
@@ -495,7 +510,7 @@ func TestAWSSDProvider_ListInstancesByService(t *testing.T) {
 		instances:  instances,
 	}
 
-	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "")
+	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
 
 	result, err := provider.ListInstancesByServiceID(services["private"]["srv1"].Id)
 	require.NoError(t, err)
@@ -532,7 +547,7 @@ func TestAWSSDProvider_CreateService(t *testing.T) {
 
 	expectedServices := make(map[string]*sd.Service)
 
-	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "")
+	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
 
 	// A type
 	provider.CreateService(aws.String("private"), aws.String("A-srv"), &endpoint.Endpoint{
@@ -636,7 +651,7 @@ func TestAWSSDProvider_UpdateService(t *testing.T) {
 		services:   services,
 	}
 
-	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "")
+	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
 
 	// update service with different TTL
 	provider.UpdateService(services["private"]["srv1"], &endpoint.Endpoint{
@@ -647,6 +662,56 @@ func TestAWSSDProvider_UpdateService(t *testing.T) {
 	assert.Equal(t, int64(100), *api.services["private"]["srv1"].DnsConfig.DnsRecords[0].TTL)
 }
 
+func TestAWSSDProvider_DeleteService(t *testing.T) {
+	namespaces := map[string]*sd.Namespace{
+		"private": {
+			Id:   aws.String("private"),
+			Name: aws.String("private.com"),
+			Type: aws.String(sd.NamespaceTypeDnsPrivate),
+		},
+	}
+
+	services := map[string]map[string]*sd.Service{
+		"private": {
+			"srv1": {
+				Id:          aws.String("srv1"),
+				Description: aws.String("heritage=external-dns,external-dns/owner=owner-id"),
+				Name:        aws.String("service1"),
+				NamespaceId: aws.String("private"),
+			},
+			"srv2": {
+				Id:          aws.String("srv2"),
+				Description: aws.String("heritage=external-dns,external-dns/owner=owner-id"),
+				Name:        aws.String("service2"),
+				NamespaceId: aws.String("private"),
+			},
+		},
+	}
+
+	api := &AWSSDClientStub{
+		namespaces: namespaces,
+		services:   services,
+	}
+
+	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "owner-id")
+
+	// delete fist service
+	err := provider.DeleteService(services["private"]["srv1"])
+	assert.NoError(t, err)
+	assert.Len(t, api.services["private"], 1)
+
+	expectedServices := map[string]*sd.Service{
+		"srv2": {
+			Id:          aws.String("srv2"),
+			Description: aws.String("heritage=external-dns,external-dns/owner=owner-id"),
+			Name:        aws.String("service2"),
+			NamespaceId: aws.String("private"),
+		},
+	}
+
+	assert.Equal(t, expectedServices, api.services["private"])
+}
+
 func TestAWSSDProvider_RegisterInstance(t *testing.T) {
 	namespaces := map[string]*sd.Namespace{
 		"private": {
@@ -703,7 +768,7 @@ func TestAWSSDProvider_RegisterInstance(t *testing.T) {
 		instances:  make(map[string]map[string]*sd.Instance),
 	}
 
-	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "")
+	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
 
 	expectedInstances := make(map[string]*sd.Instance)
 
@@ -820,7 +885,7 @@ func TestAWSSDProvider_DeregisterInstance(t *testing.T) {
 		instances:  instances,
 	}
 
-	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "")
+	provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
 
 	provider.DeregisterInstance(services["private"]["srv1"], endpoint.NewEndpoint("srv1.private.com.", endpoint.RecordTypeA, "1.2.3.4"))