Implementation of AWS ServiceDiscovery provider

Gopkg.lock

@@ -82,12 +82,15 @@
     "internal/sdkrand",
     "internal/shareddefaults",
     "private/protocol",
+    "private/protocol/json/jsonutil",
+    "private/protocol/jsonrpc",
     "private/protocol/query",
     "private/protocol/query/queryutil",
     "private/protocol/rest",
     "private/protocol/restxml",
     "private/protocol/xml/xmlutil",
     "service/route53",
+    "service/servicediscovery",
     "service/sts"
   ]
   revision = "9b0098a71f6d4d473a26ec8ad3c2feaac6eb1da6"
@@ -645,6 +648,6 @@
 [solve-meta]
   analyzer-name = "dep"
   analyzer-version = 1
-  inputs-digest = "fe45c175af381b4250e119e8683aaa1d25c69461c58d9df7012c97afbc7caa9d"
+  inputs-digest = "d5deea43eb04e9ef3a6ecb3589b91c149e092505f66905baa01c67379776d231"
   solver-name = "gps-cdcl"
   solver-version = 1

README.md

@@ -45,7 +45,8 @@ Note that all flags can be replaced with environment variables; for instance,
 
 The following tutorials are provided:
 
-* [AWS](docs/tutorials/aws.md)
+* [AWS (Route53)](docs/tutorials/aws.md)
+* [AWS (Service Discovery)](docs/tutorials/aws-sd.md)
 * [Azure](docs/tutorials/azure.md)
 * [Cloudflare](docs/tutorials/cloudflare.md)
 * [DigitalOcean](docs/tutorials/digitalocean.md)

docs/tutorials/aws-sd.md

@@ -0,0 +1,200 @@
+# Setting up ExternalDNS using AWS Service Discovery API
+
+This tutorial describes how to set up ExternalDNS for usage within a Kubernetes cluster on AWS with [Service Discovery API](https://docs.aws.amazon.com/Route53/latest/APIReference/overview-service-discovery.html).
+
+The **Service Discovery API** is an alternative approach to managing DNS records directly using the Route53 API. It is more suitable for a dynamic environment where service endpoints change frequently. It abstracts away technical details of the DNS protocol and offers a simplified model. Service discovery consists of three main API calls:
+
+* CreatePublicDnsNamespace – automatically creates a DNS hosted zone
+* CreateService – creates a new named service inside the specified namespace
+* RegisterInstance/DeregisterInstance – can be called multiple times to create a DNS record for the specified *Service*
+
+Learn more about the API in the [Amazon Route 53 API Reference](https://docs.aws.amazon.com/Route53/latest/APIReference/API_Operations_Amazon_Route_53_Auto_Naming.html).
+
+
+## IAM Permissions
+
+To use the service discovery API, a user executing the ExternalDNS must have the permissions in the `AmazonRoute53AutoNamingFullAccess` managed policy.
+
+```
+{
+    "Version": "2012-10-17",
+    "Statement": [
+        {
+            "Effect": "Allow",
+            "Action": [
+                "route53:GetHostedZone",
+                "route53:ListHostedZonesByName",
+                "route53:CreateHostedZone",
+                "route53:DeleteHostedZone",
+                "route53:ChangeResourceRecordSets",
+                "route53:CreateHealthCheck",
+                "route53:GetHealthCheck",
+                "route53:DeleteHealthCheck",
+                "route53:UpdateHealthCheck",
+                "ec2:DescribeVpcs",
+                "ec2:DescribeRegions",
+                "servicediscovery:*"
+            ],
+            "Resource": [
+                "*"
+            ]
+        }
+    ]
+}
+```
+
+## Set up a namespace
+
+Create a DNS namespace using the service discovery API
+
+```console
+$ aws servicediscovery create-public-dns-namespace --name "external-dns-test.my-org.com"
+```
+
+Verify that the namespace was truly created
+
+```console
+$ aws servicediscovery list-namespaces
+```
+
+## Deploy ExternalDNS
+
+Connect your `kubectl` client to the cluster that you want to test ExternalDNS with.
+Then apply the following manifest file to deploy ExternalDNS.
+
+```yaml
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: external-dns
+spec:
+  strategy:
+    type: Recreate
+  template:
+    metadata:
+      labels:
+        app: external-dns
+    spec:
+      containers:
+      - name: external-dns
+        image: registry.opensource.zalan.do/teapot/external-dns:v0.4.8
+        args:
+        - --source=service
+        - --source=ingress
+        - --domain-filter=external-dns-test.my-org.com # Makes ExternalDNS see only the namespaces that match the specified domain. Omit the filter if you want to process all available namespaces.
+        - --provider=aws-sd
+        - --aws-zone-type=public # Only look at public namespaces. Valid values are public, private, or no value for both)
+        - --txt-owner-id=my-identifier
+```
+
+
+## Verify that ExternalDNS works (Service example)
+
+Create the following sample application to test that ExternalDNS works.
+
+> For services ExternalDNS will look for the annotation `external-dns.alpha.kubernetes.io/hostname` on the service and use the corresponding value.
+
+```yaml
+apiVersion: v1
+kind: Service
+metadata:
+  name: nginx
+  annotations:
+    external-dns.alpha.kubernetes.io/hostname: nginx.external-dns-test.my-org.com
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 80
+    name: http
+    targetPort: 80
+  selector:
+    app: nginx
+
+---
+
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: nginx
+spec:
+  template:
+    metadata:
+      labels:
+        app: nginx
+    spec:
+      containers:
+      - image: nginx
+        name: nginx
+        ports:
+        - containerPort: 80
+          name: http
+```
+
+After one minute check that a corresponding DNS record for your service was created in your hosted zone. We recommended that you use the [Amazon Route53 console](https://console.aws.amazon.com/route53) for that purpose.
+
+
+## Custom TTL
+
+The default DNS record TTL (time to live) is 300 seconds. You can customize this value by setting the annotation `external-dns.alpha.kubernetes.io/ttl`.
+For example, modify the service manifest YAML file above:
+
+```yaml
+apiVersion: v1
+kind: Service
+metadata:
+  name: nginx
+  annotations:
+    external-dns.alpha.kubernetes.io/hostname: nginx.external-dns-test.my-org.com
+    external-dns.alpha.kubernetes.io/ttl: 60
+spec:
+    ...
+```
+
+This will set the TTL for the DNS record to 60 seconds.
+
+
+## Clean up
+
+Delete all service objects before terminating the cluster so all load balancers get cleaned up correctly.
+
+```console
+$ kubectl delete service nginx
+```
+
+Give ExternalDNS some time to clean up the DNS records for you. Then delete the remaining service and namespace.
+
+```console
+$ aws servicediscovery list-services
+
+{
+    "Services": [
+        {
+            "Id": "srv-6dygt5ywvyzvi3an",
+            "Arn": "arn:aws:servicediscovery:us-west-2:861574988794:service/srv-6dygt5ywvyzvi3an",
+            "Name": "nginx"
+        }
+    ]
+}
+```
+
+```console
+$ aws servicediscovery delete-service --id srv-6dygt5ywvyzvi3an
+```
+
+```console
+$ aws servicediscovery list-namespaces
+{
+    "Namespaces": [
+        {
+            "Type": "DNS_PUBLIC",
+            "Id": "ns-durf2oxu4gxcgo6z",
+            "Arn": "arn:aws:servicediscovery:us-west-2:861574988794:namespace/ns-durf2oxu4gxcgo6z",
+            "Name": "external-dns-test.my-org.com"
+        }
+    ]
+}
+```
+
+```console
+$ aws servicediscovery delete-namespace --id ns-durf2oxu4gxcgo6z
+```

endpoint/labels.go

@@ -34,6 +34,10 @@ const (
 	OwnerLabelKey = "owner"
 	// ResourceLabelKey is the name of the label that identifies k8s resource which wants to acquire the DNS name
 	ResourceLabelKey = "resource"
+
+	// AWSSDDescriptionLabel label responsible for storing raw owner/resource combination information in the Labels
+	// supposed to be inserted by AWS SD Provider, and parsed into OwnerLabelKey and ResourceLabelKey key by AWS SD Registry
+	AWSSDDescriptionLabel = "aws-sd-description"
 )
 
 // Labels store metadata related to the endpoint

main.go

@@ -95,6 +95,13 @@ func main() {
 	switch cfg.Provider {
 	case "aws":
 		p, err = provider.NewAWSProvider(domainFilter, zoneIDFilter, zoneTypeFilter, cfg.AWSAssumeRole, cfg.DryRun)
+	case "aws-sd":
+		// Check that only compatible Registry is used with AWS-SD
+		if cfg.Registry != "noop" && cfg.Registry != "aws-sd-registry" {
+			log.Infof("Registry \"%s\" cannot be used with AWS ServiceDiscovery. Switching to \"aws-sd-registry\".", cfg.Registry)
+			cfg.Registry = "aws-sd-registry"
+		}
+		p, err = provider.NewAWSSDProvider(domainFilter, cfg.AWSZoneType, cfg.DryRun)
 	case "azure":
 		p, err = provider.NewAzureProvider(cfg.AzureConfigFile, domainFilter, zoneIDFilter, cfg.AzureResourceGroup, cfg.DryRun)
 	case "cloudflare":
@@ -151,6 +158,8 @@ func main() {
 		r, err = registry.NewNoopRegistry(p)
 	case "txt":
 		r, err = registry.NewTXTRegistry(p, cfg.TXTPrefix, cfg.TXTOwnerID)
+	case "aws-sd-registry":
+		r, err = registry.NewAWSSDRegistry(p.(*provider.AWSSDProvider), cfg.TXTOwnerID)
 	default:
 		log.Fatalf("unknown registry: %s", cfg.Registry)
 	}
@@ -179,7 +188,6 @@ func main() {
 
 		os.Exit(0)
 	}
-
 	ctrl.Run(stopChan)
 }
 

pkg/apis/externaldns/types.go

@@ -165,7 +165,7 @@ func (cfg *Config) ParseFlags(args []string) error {
 	app.Flag("publish-internal-services", "Allow external-dns to publish DNS records for ClusterIP services (optional)").BoolVar(&cfg.PublishInternal)
 
 	// Flags related to providers
-	app.Flag("provider", "The DNS provider where the DNS records will be created (required, options: aws, google, azure, cloudflare, digitalocean, dnsimple, infoblox, dyn, designate, inmemory, pdns)").Required().PlaceHolder("provider").EnumVar(&cfg.Provider, "aws", "google", "azure", "cloudflare", "digitalocean", "dnsimple", "infoblox", "dyn", "designate", "inmemory", "pdns")
+	app.Flag("provider", "The DNS provider where the DNS records will be created (required, options: aws, aws-sd, google, azure, cloudflare, digitalocean, dnsimple, infoblox, dyn, designate, inmemory, pdns)").Required().PlaceHolder("provider").EnumVar(&cfg.Provider, "aws", "aws-sd", "google", "azure", "cloudflare", "digitalocean", "dnsimple", "infoblox", "dyn", "designate", "inmemory", "pdns")
 	app.Flag("domain-filter", "Limit possible target zones by a domain suffix; specify multiple times for multiple domains (optional)").Default("").StringsVar(&cfg.DomainFilter)
 	app.Flag("zone-id-filter", "Filter target zones by hosted zone id; specify multiple times for multiple zones (optional)").Default("").StringsVar(&cfg.ZoneIDFilter)
 	app.Flag("google-project", "When using the Google provider, current project is auto-detected, when running on GCP. Specify other project with this. Must be specified when running outside GCP.").Default(defaultConfig.GoogleProject).StringVar(&cfg.GoogleProject)
@@ -193,7 +193,7 @@ func (cfg *Config) ParseFlags(args []string) error {
 	app.Flag("policy", "Modify how DNS records are sychronized between sources and providers (default: sync, options: sync, upsert-only)").Default(defaultConfig.Policy).EnumVar(&cfg.Policy, "sync", "upsert-only")
 
 	// Flags related to the registry
-	app.Flag("registry", "The registry implementation to use to keep track of DNS record ownership (default: txt, options: txt, noop)").Default(defaultConfig.Registry).EnumVar(&cfg.Registry, "txt", "noop")
+	app.Flag("registry", "The registry implementation to use to keep track of DNS record ownership (default: txt, options: txt, noop, aws-sd-registry)").Default(defaultConfig.Registry).EnumVar(&cfg.Registry, "txt", "noop", "aws-sd-registry")
 	app.Flag("txt-owner-id", "When using the TXT registry, a name that identifies this instance of ExternalDNS (default: default)").Default(defaultConfig.TXTOwnerID).StringVar(&cfg.TXTOwnerID)
 	app.Flag("txt-prefix", "When using the TXT registry, a custom string that's prefixed to each ownership DNS record (optional)").Default(defaultConfig.TXTPrefix).StringVar(&cfg.TXTPrefix)
 

provider/aws_sd.go

@@ -0,0 +1,662 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package provider
+
+import (
+	"strings"
+
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/request"
+	"github.com/aws/aws-sdk-go/aws/session"
+	sd "github.com/aws/aws-sdk-go/service/servicediscovery"
+	"github.com/kubernetes-incubator/external-dns/endpoint"
+	"github.com/kubernetes-incubator/external-dns/pkg/apis/externaldns"
+	"github.com/kubernetes-incubator/external-dns/plan"
+	"github.com/linki/instrumented_http"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	sdElbHostnameSuffix = ".elb.amazonaws.com"
+	sdDefaultRecordTTL  = 300
+
+	sdNamespaceTypePublic  = "public"
+	sdNamespaceTypePrivate = "private"
+
+	sdInstanceAttrIPV4  = "AWS_INSTANCE_IPV4"
+	sdInstanceAttrCname = "AWS_INSTANCE_CNAME"
+	sdInstanceAttrAlias = "AWS_ALIAS_DNS_NAME"
+)
+
+// AWSSDClient is the subset of the AWS Route53 Auto Naming API that we actually use. Add methods as required.
+// Signatures must match exactly. Taken from https://github.com/aws/aws-sdk-go/blob/master/service/servicediscovery/api.go
+type AWSSDClient interface {
+	CreateService(input *sd.CreateServiceInput) (*sd.CreateServiceOutput, error)
+	DeregisterInstance(input *sd.DeregisterInstanceInput) (*sd.DeregisterInstanceOutput, error)
+	GetService(input *sd.GetServiceInput) (*sd.GetServiceOutput, error)
+	ListInstancesPages(input *sd.ListInstancesInput, fn func(*sd.ListInstancesOutput, bool) bool) error
+	ListNamespacesPages(input *sd.ListNamespacesInput, fn func(*sd.ListNamespacesOutput, bool) bool) error
+	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)
+}
+
+// AWSSDProvider is an implementation of Provider for AWS Route53 Auto Naming.
+type AWSSDProvider struct {
+	client AWSSDClient
+	dryRun bool
+	// only consider namespaces ending in this suffix
+	namespaceFilter DomainFilter
+	// filter namespace by type (private or public)
+	namespaceTypeFilter *sd.NamespaceFilter
+}
+
+// NewAWSSDProvider initializes a new AWS Route53 Auto Naming based Provider.
+func NewAWSSDProvider(domainFilter DomainFilter, namespaceType string, dryRun bool) (*AWSSDProvider, error) {
+	config := aws.NewConfig()
+
+	config = config.WithHTTPClient(
+		instrumented_http.NewClient(config.HTTPClient, &instrumented_http.Callbacks{
+			PathProcessor: func(path string) string {
+				parts := strings.Split(path, "/")
+				return parts[len(parts)-1]
+			},
+		}),
+	)
+
+	sess, err := session.NewSessionWithOptions(session.Options{
+		Config:            *config,
+		SharedConfigState: session.SharedConfigEnable,
+	})
+	if err != nil {
+		return nil, err
+	}
+	sess.Handlers.Build.PushBack(request.MakeAddToUserAgentHandler("ExternalDNS", externaldns.Version))
+
+	provider := &AWSSDProvider{
+		client:              sd.New(sess),
+		namespaceFilter:     domainFilter,
+		namespaceTypeFilter: newSdNamespaceFilter(namespaceType),
+		dryRun:              dryRun,
+	}
+
+	return provider, nil
+}
+
+// newSdNamespaceFilter initialized AWS SD Namespace Filter based on given string config
+func newSdNamespaceFilter(namespaceTypeConfig string) *sd.NamespaceFilter {
+	switch namespaceTypeConfig {
+	case sdNamespaceTypePublic:
+		return &sd.NamespaceFilter{
+			Name:   aws.String(sd.NamespaceFilterNameType),
+			Values: []*string{aws.String(sd.NamespaceTypeDnsPublic)},
+		}
+	case sdNamespaceTypePrivate:
+		return &sd.NamespaceFilter{
+			Name:   aws.String(sd.NamespaceFilterNameType),
+			Values: []*string{aws.String(sd.NamespaceTypeDnsPrivate)},
+		}
+	default:
+		return nil
+	}
+}
+
+// Records returns list of all endpoints.
+func (p *AWSSDProvider) Records() (endpoints []*endpoint.Endpoint, err error) {
+	namespaces, err := p.ListNamespaces()
+	if err != nil {
+		return nil, err
+	}
+
+	for _, ns := range namespaces {
+		services, err := p.ListServicesByNamespaceID(ns.Id)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, srv := range services {
+			instances, err := p.ListInstancesByServiceID(srv.Id)
+			if err != nil {
+				return nil, err
+			}
+
+			if len(instances) > 0 {
+				ep := p.instancesToEndpoint(ns, srv, instances)
+				endpoints = append(endpoints, ep)
+			}
+		}
+	}
+
+	return endpoints, nil
+}
+
+func (p *AWSSDProvider) instancesToEndpoint(ns *sd.NamespaceSummary, srv *sd.Service, instances []*sd.InstanceSummary) *endpoint.Endpoint {
+	// DNS name of the record is a concatenation of service and namespace
+	recordName := *srv.Name + "." + *ns.Name
+
+	labels := endpoint.NewLabels()
+	labels[endpoint.AWSSDDescriptionLabel] = aws.StringValue(srv.Description)
+
+	newEndpoint := &endpoint.Endpoint{
+		DNSName:   recordName,
+		RecordTTL: endpoint.TTL(aws.Int64Value(srv.DnsConfig.DnsRecords[0].TTL)),
+		Targets:   make(endpoint.Targets, 0, len(instances)),
+		Labels:    labels,
+	}
+
+	for _, inst := range instances {
+		// CNAME
+		if inst.Attributes[sdInstanceAttrCname] != nil && aws.StringValue(srv.DnsConfig.DnsRecords[0].Type) == sd.RecordTypeCname {
+			newEndpoint.RecordType = endpoint.RecordTypeCNAME
+			newEndpoint.Targets = append(newEndpoint.Targets, aws.StringValue(inst.Attributes[sdInstanceAttrCname]))
+
+			// ALIAS
+		} else if inst.Attributes[sdInstanceAttrAlias] != nil {
+			newEndpoint.RecordType = endpoint.RecordTypeCNAME
+			newEndpoint.Targets = append(newEndpoint.Targets, aws.StringValue(inst.Attributes[sdInstanceAttrAlias]))
+
+			// IP-based target
+		} else if inst.Attributes[sdInstanceAttrIPV4] != nil {
+			newEndpoint.RecordType = endpoint.RecordTypeA
+			newEndpoint.Targets = append(newEndpoint.Targets, aws.StringValue(inst.Attributes[sdInstanceAttrIPV4]))
+		} else {
+			log.Warnf("Invalid instance \"%v\" found in service \"%v\"", inst, srv.Name)
+		}
+	}
+
+	return newEndpoint
+}
+
+// ApplyChanges applies Kubernetes changes in endpoints to AWS API
+func (p *AWSSDProvider) ApplyChanges(changes *plan.Changes) error {
+	// return early if there is nothing to change
+	if len(changes.Create) == 0 && len(changes.Delete) == 0 && len(changes.UpdateNew) == 0 {
+		log.Info("All records are already up to date")
+		return nil
+	}
+
+	// convert updates to delete and create operation if applicable (updates not supported)
+	creates, deletes := p.updatesToCreates(changes)
+	changes.Delete = append(changes.Delete, deletes...)
+	changes.Create = append(changes.Create, creates...)
+
+	namespaces, err := p.ListNamespaces()
+	if err != nil {
+		return err
+	}
+
+	// Deletes must be executed first to support update case.
+	// When just list of targets is updated `[1.2.3.4] -> [1.2.3.4, 1.2.3.5]` it is translated to:
+	// ```
+	// deletes = [1.2.3.4]
+	// creates = [1.2.3.4, 1.2.3.5]
+	// ```
+	// then when deletes are executed after creates it will miss the `1.2.3.4` instance.
+	err = p.submitDeletes(namespaces, changes.Delete)
+	if err != nil {
+		return err
+	}
+
+	err = p.submitCreates(namespaces, changes.Create)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (p *AWSSDProvider) updatesToCreates(changes *plan.Changes) (creates []*endpoint.Endpoint, deletes []*endpoint.Endpoint) {
+	updateNewMap := map[string]*endpoint.Endpoint{}
+	for _, e := range changes.UpdateNew {
+		updateNewMap[e.DNSName] = e
+	}
+
+	for _, old := range changes.UpdateOld {
+		current := updateNewMap[old.DNSName]
+
+		if !old.Targets.Same(current.Targets) {
+			// when targets differ the old instances need to be de-registered first
+			deletes = append(deletes, old)
+		}
+
+		// always register (or re-register) instance with the current data
+		creates = append(creates, current)
+	}
+
+	return creates, deletes
+}
+
+func (p *AWSSDProvider) submitCreates(namespaces []*sd.NamespaceSummary, changes []*endpoint.Endpoint) error {
+	changesByNamespaceID := p.changesByNamespaceID(namespaces, changes)
+
+	for nsID, changeList := range changesByNamespaceID {
+		services, err := p.ListServicesByNamespaceID(aws.String(nsID))
+		if err != nil {
+			return err
+		}
+
+		for _, ch := range changeList {
+			_, srvName := p.parseHostname(ch.DNSName)
+
+			srv := services[srvName]
+			if srv == nil {
+				// when service is missing create a new one
+				srv, err = p.CreateService(&nsID, &srvName, ch)
+				if err != nil {
+					return err
+				}
+				// update local list of services
+				services[*srv.Name] = srv
+			} else {
+				// update service when TTL or Description differ
+				if (ch.RecordTTL.IsConfigured() && *srv.DnsConfig.DnsRecords[0].TTL != int64(ch.RecordTTL)) ||
+					aws.StringValue(srv.Description) != ch.Labels[endpoint.AWSSDDescriptionLabel] {
+					err = p.UpdateService(srv, ch)
+					if err != nil {
+						return err
+					}
+				}
+			}
+
+			err = p.RegisterInstance(srv, ch)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+func (p *AWSSDProvider) submitDeletes(namespaces []*sd.NamespaceSummary, changes []*endpoint.Endpoint) error {
+	changesByNamespaceID := p.changesByNamespaceID(namespaces, changes)
+
+	for nsID, changeList := range changesByNamespaceID {
+		services, err := p.ListServicesByNamespaceID(aws.String(nsID))
+		if err != nil {
+			return err
+		}
+
+		for _, ch := range changeList {
+			hostname := ch.DNSName
+			_, srvName := p.parseHostname(hostname)
+
+			srv := services[srvName]
+			if srv == nil {
+				return fmt.Errorf("service \"%s\" is missing when trying to delete \"%v\"", srvName, hostname)
+			}
+
+			err := p.DeregisterInstance(srv, ch)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+// ListNamespaces returns all namespaces matching defined namespace filter
+func (p *AWSSDProvider) ListNamespaces() ([]*sd.NamespaceSummary, error) {
+	namespaces := make([]*sd.NamespaceSummary, 0)
+
+	f := func(resp *sd.ListNamespacesOutput, lastPage bool) bool {
+		for _, ns := range resp.Namespaces {
+			if !p.namespaceFilter.Match(aws.StringValue(ns.Name)) {
+				continue
+			}
+			namespaces = append(namespaces, ns)
+		}
+
+		return true
+	}
+
+	err := p.client.ListNamespacesPages(&sd.ListNamespacesInput{
+		Filters: []*sd.NamespaceFilter{p.namespaceTypeFilter},
+	}, f)
+	if err != nil {
+		return nil, err
+	}
+
+	return namespaces, nil
+}
+
+// ListServicesByNamespaceID returns list of services in given namespace. Returns map[srv_name]*sd.Service
+func (p *AWSSDProvider) ListServicesByNamespaceID(namespaceID *string) (map[string]*sd.Service, error) {
+	serviceIds := make([]*string, 0)
+
+	f := func(resp *sd.ListServicesOutput, lastPage bool) bool {
+		for _, srv := range resp.Services {
+			serviceIds = append(serviceIds, srv.Id)
+		}
+
+		return true
+	}
+
+	err := p.client.ListServicesPages(&sd.ListServicesInput{
+		Filters: []*sd.ServiceFilter{{
+			Name:   aws.String(sd.ServiceFilterNameNamespaceId),
+			Values: []*string{namespaceID},
+		}},
+	}, f)
+	if err != nil {
+		return nil, err
+	}
+
+	// get detail of each listed service
+	services := make(map[string]*sd.Service)
+	for _, serviceID := range serviceIds {
+		service, err := p.GetServiceDetail(serviceID)
+		if err != nil {
+			return nil, err
+		}
+
+		services[aws.StringValue(service.Name)] = service
+	}
+
+	return services, nil
+}
+
+// GetServiceDetail returns detail of given service
+func (p *AWSSDProvider) GetServiceDetail(serviceID *string) (*sd.Service, error) {
+	output, err := p.client.GetService(&sd.GetServiceInput{
+		Id: serviceID,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return output.Service, nil
+}
+
+// ListInstancesByServiceID returns list of instances registered in given service.
+func (p *AWSSDProvider) ListInstancesByServiceID(serviceID *string) ([]*sd.InstanceSummary, error) {
+	instances := make([]*sd.InstanceSummary, 0)
+
+	f := func(resp *sd.ListInstancesOutput, lastPage bool) bool {
+		instances = append(instances, resp.Instances...)
+
+		return true
+	}
+
+	err := p.client.ListInstancesPages(&sd.ListInstancesInput{
+		ServiceId: serviceID,
+	}, f)
+	if err != nil {
+		return nil, err
+	}
+
+	return instances, nil
+}
+
+// CreateService creates a new service in AWS API. Returns the created service.
+func (p *AWSSDProvider) CreateService(namespaceID *string, srvName *string, ep *endpoint.Endpoint) (*sd.Service, error) {
+	log.Infof("Creating a new service \"%s\" in \"%s\" namespace", *srvName, *namespaceID)
+
+	srvType := p.serviceTypeFromEndpoint(ep)
+	routingPolicy := p.routingPolicyFromEndpoint(ep)
+
+	ttl := int64(sdDefaultRecordTTL)
+	if ep.RecordTTL.IsConfigured() {
+		ttl = int64(ep.RecordTTL)
+	}
+
+	if !p.dryRun {
+		out, err := p.client.CreateService(&sd.CreateServiceInput{
+			Name:        srvName,
+			Description: aws.String(ep.Labels[endpoint.AWSSDDescriptionLabel]),
+			DnsConfig: &sd.DnsConfig{
+				NamespaceId:   namespaceID,
+				RoutingPolicy: aws.String(routingPolicy),
+				DnsRecords: []*sd.DnsRecord{{
+					Type: aws.String(srvType),
+					TTL:  aws.Int64(ttl),
+				}},
+			},
+		})
+		if err != nil {
+			return nil, err
+		}
+
+		return out.Service, nil
+	}
+
+	// return mock service summary in case of dry run
+	return &sd.Service{Id: aws.String("dry-run-service"), Name: aws.String("dry-run-service")}, nil
+}
+
+// UpdateService updates the specified service with information from provided endpoint.
+func (p *AWSSDProvider) UpdateService(service *sd.Service, ep *endpoint.Endpoint) error {
+	log.Infof("Updating service \"%s\"", *service.Name)
+
+	srvType := p.serviceTypeFromEndpoint(ep)
+
+	ttl := int64(sdDefaultRecordTTL)
+	if ep.RecordTTL.IsConfigured() {
+		ttl = int64(ep.RecordTTL)
+	}
+
+	if !p.dryRun {
+		_, err := p.client.UpdateService(&sd.UpdateServiceInput{
+			Id: service.Id,
+			Service: &sd.ServiceChange{
+				Description: aws.String(ep.Labels[endpoint.AWSSDDescriptionLabel]),
+				DnsConfig: &sd.DnsConfigChange{
+					DnsRecords: []*sd.DnsRecord{{
+						Type: aws.String(srvType),
+						TTL:  aws.Int64(ttl),
+					}},
+				}}})
+		if err != nil {
+			return err
+		}
+	}
+
+	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 {
+		log.Infof("Registering a new instance \"%s\" for service \"%s\" (%s)", target, *service.Name, *service.Id)
+
+		attr := make(map[string]*string)
+
+		if ep.RecordType == endpoint.RecordTypeCNAME {
+			if p.isAWSLoadBalancer(target) {
+				attr[sdInstanceAttrAlias] = aws.String(target)
+			} else {
+				attr[sdInstanceAttrCname] = aws.String(target)
+			}
+		} else if ep.RecordType == endpoint.RecordTypeA {
+			attr[sdInstanceAttrIPV4] = aws.String(target)
+		} else {
+			return fmt.Errorf("invalid endpoint type (%v)", ep)
+		}
+
+		if !p.dryRun {
+			_, err := p.client.RegisterInstance(&sd.RegisterInstanceInput{
+				ServiceId:  service.Id,
+				Attributes: attr,
+				InstanceId: aws.String(p.targetToInstanceID(target)),
+			})
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+// DeregisterInstance removes an instance from given service.
+func (p *AWSSDProvider) DeregisterInstance(service *sd.Service, ep *endpoint.Endpoint) error {
+	for _, target := range ep.Targets {
+		log.Infof("De-registering an instance \"%s\" for service \"%s\" (%s)", target, *service.Name, *service.Id)
+
+		if !p.dryRun {
+			_, err := p.client.DeregisterInstance(&sd.DeregisterInstanceInput{
+				InstanceId: aws.String(p.targetToInstanceID(target)),
+				ServiceId:  service.Id,
+			})
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+// Instance ID length is limited by AWS API to 64 characters. For longer strings SHA-256 hash will be used instead of
+// the verbatim target to limit the length.
+func (p *AWSSDProvider) targetToInstanceID(target string) string {
+	if len(target) > 64 {
+		hash := sha256.Sum256([]byte(strings.ToLower(target)))
+		return hex.EncodeToString(hash[:])
+	}
+
+	return strings.ToLower(target)
+}
+
+// nolint: deadcode
+// used from unit test
+func namespaceToNamespaceSummary(namespace *sd.Namespace) *sd.NamespaceSummary {
+	if namespace == nil {
+		return nil
+	}
+
+	return &sd.NamespaceSummary{
+		Id:   namespace.Id,
+		Type: namespace.Type,
+		Name: namespace.Name,
+		Arn:  namespace.Arn,
+	}
+}
+
+// nolint: deadcode
+// used from unit test
+func serviceToServiceSummary(service *sd.Service) *sd.ServiceSummary {
+	if service == nil {
+		return nil
+	}
+
+	return &sd.ServiceSummary{
+		Name:          service.Name,
+		Id:            service.Id,
+		Arn:           service.Arn,
+		Description:   service.Description,
+		InstanceCount: service.InstanceCount,
+	}
+}
+
+// nolint: deadcode
+// used from unit test
+func instanceToInstanceSummary(instance *sd.Instance) *sd.InstanceSummary {
+	if instance == nil {
+		return nil
+	}
+
+	return &sd.InstanceSummary{
+		Id:         instance.Id,
+		Attributes: instance.Attributes,
+	}
+}
+
+func (p *AWSSDProvider) changesByNamespaceID(namespaces []*sd.NamespaceSummary, changes []*endpoint.Endpoint) map[string][]*endpoint.Endpoint {
+	changesByNsID := make(map[string][]*endpoint.Endpoint)
+
+	for _, ns := range namespaces {
+		changesByNsID[*ns.Id] = []*endpoint.Endpoint{}
+	}
+
+	for _, c := range changes {
+		// trim the trailing dot from hostname if any
+		hostname := strings.TrimSuffix(c.DNSName, ".")
+		nsName, _ := p.parseHostname(hostname)
+
+		matchingNamespaces := matchingNamespaces(nsName, namespaces)
+		if len(matchingNamespaces) == 0 {
+			log.Warnf("Skipping record %s because no namespace matching record DNS Name was detected ", c.String())
+			continue
+		}
+		for _, ns := range matchingNamespaces {
+			changesByNsID[*ns.Id] = append(changesByNsID[*ns.Id], c)
+		}
+	}
+
+	// separating a change could lead to empty sub changes, remove them here.
+	for zone, change := range changesByNsID {
+		if len(change) == 0 {
+			delete(changesByNsID, zone)
+		}
+	}
+
+	return changesByNsID
+}
+
+// returns list of all namespaces matching given hostname
+func matchingNamespaces(hostname string, namespaces []*sd.NamespaceSummary) []*sd.NamespaceSummary {
+	matchingNamespaces := make([]*sd.NamespaceSummary, 0)
+
+	for _, ns := range namespaces {
+		if *ns.Name == hostname {
+			matchingNamespaces = append(matchingNamespaces, ns)
+		}
+	}
+
+	return matchingNamespaces
+}
+
+// parse hostname to namespace (domain) and service
+func (p *AWSSDProvider) parseHostname(hostname string) (namespace string, service string) {
+	parts := strings.Split(hostname, ".")
+	service = parts[0]
+	namespace = strings.Join(parts[1:], ".")
+	return
+}
+
+// determine service routing policy based on endpoint type
+func (p *AWSSDProvider) routingPolicyFromEndpoint(ep *endpoint.Endpoint) string {
+	if ep.RecordType == endpoint.RecordTypeA {
+		return sd.RoutingPolicyMultivalue
+	}
+
+	return sd.RoutingPolicyWeighted
+}
+
+// determine service type (A, CNAME) from given endpoint
+func (p *AWSSDProvider) serviceTypeFromEndpoint(ep *endpoint.Endpoint) string {
+	if ep.RecordType == endpoint.RecordTypeCNAME {
+		// FIXME service type is derived from the first target only. Theoretically this may be problem.
+		// But I don't see a scenario where one endpoint contains targets of different types.
+		if p.isAWSLoadBalancer(ep.Targets[0]) {
+			// ALIAS target uses DNS record type of A
+			return sd.RecordTypeA
+		}
+		return sd.RecordTypeCname
+	}
+	return sd.RecordTypeA
+}
+
+// determine if a given hostname belongs to an AWS load balancer
+func (p *AWSSDProvider) isAWSLoadBalancer(hostname string) bool {
+	return strings.HasSuffix(hostname, sdElbHostnameSuffix)
+}

provider/aws_sd_test.go

@@ -0,0 +1,808 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package provider
+
+import (
+	"errors"
+	"math/rand"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	sd "github.com/aws/aws-sdk-go/service/servicediscovery"
+	"github.com/kubernetes-incubator/external-dns/endpoint"
+	"github.com/kubernetes-incubator/external-dns/internal/testutils"
+	"github.com/kubernetes-incubator/external-dns/plan"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Compile time check for interface conformance
+var _ AWSSDClient = &AWSSDClientStub{}
+
+type AWSSDClientStub struct {
+	// map[namespace_id]namespace
+	namespaces map[string]*sd.Namespace
+
+	// map[namespace_id] => map[service_id]instance
+	services map[string]map[string]*sd.Service
+
+	// map[service_id] => map[inst_id]instance
+	instances map[string]map[string]*sd.Instance
+}
+
+func (s *AWSSDClientStub) CreateService(input *sd.CreateServiceInput) (*sd.CreateServiceOutput, error) {
+
+	srv := &sd.Service{
+		Id:               aws.String(string(rand.Intn(10000))),
+		DnsConfig:        input.DnsConfig,
+		Name:             input.Name,
+		Description:      input.Description,
+		CreateDate:       aws.Time(time.Now()),
+		CreatorRequestId: input.CreatorRequestId,
+	}
+
+	nsServices, ok := s.services[*input.DnsConfig.NamespaceId]
+	if !ok {
+		nsServices = make(map[string]*sd.Service)
+		s.services[*input.DnsConfig.NamespaceId] = nsServices
+	}
+	nsServices[*srv.Id] = srv
+
+	return &sd.CreateServiceOutput{
+		Service: srv,
+	}, nil
+}
+
+func (s *AWSSDClientStub) DeregisterInstance(input *sd.DeregisterInstanceInput) (*sd.DeregisterInstanceOutput, error) {
+	serviceInstances := s.instances[*input.ServiceId]
+	delete(serviceInstances, *input.InstanceId)
+
+	return &sd.DeregisterInstanceOutput{}, nil
+}
+
+func (s *AWSSDClientStub) GetService(input *sd.GetServiceInput) (*sd.GetServiceOutput, error) {
+	for _, entry := range s.services {
+		srv, ok := entry[*input.Id]
+		if ok {
+			return &sd.GetServiceOutput{
+				Service: srv,
+			}, nil
+		}
+	}
+
+	return nil, errors.New("service not found")
+}
+
+func (s *AWSSDClientStub) ListInstancesPages(input *sd.ListInstancesInput, fn func(*sd.ListInstancesOutput, bool) bool) error {
+	instances := make([]*sd.InstanceSummary, 0)
+
+	for _, inst := range s.instances[*input.ServiceId] {
+		instances = append(instances, instanceToInstanceSummary(inst))
+	}
+
+	fn(&sd.ListInstancesOutput{
+		Instances: instances,
+	}, true)
+
+	return nil
+}
+
+func (s *AWSSDClientStub) ListNamespacesPages(input *sd.ListNamespacesInput, fn func(*sd.ListNamespacesOutput, bool) bool) error {
+	namespaces := make([]*sd.NamespaceSummary, 0)
+
+	filter := input.Filters[0]
+
+	for _, ns := range s.namespaces {
+		if filter != nil && *filter.Name == sd.NamespaceFilterNameType {
+			if *ns.Type != *filter.Values[0] {
+				// skip namespaces not matching filter
+				continue
+			}
+		}
+		namespaces = append(namespaces, namespaceToNamespaceSummary(ns))
+	}
+
+	fn(&sd.ListNamespacesOutput{
+		Namespaces: namespaces,
+	}, true)
+
+	return nil
+}
+
+func (s *AWSSDClientStub) ListServicesPages(input *sd.ListServicesInput, fn func(*sd.ListServicesOutput, bool) bool) error {
+	services := make([]*sd.ServiceSummary, 0)
+
+	// get namespace filter
+	filter := input.Filters[0]
+	if filter == nil || *filter.Name != sd.ServiceFilterNameNamespaceId {
+		return errors.New("missing namespace filter")
+	}
+	nsID := filter.Values[0]
+
+	for _, srv := range s.services[*nsID] {
+		services = append(services, serviceToServiceSummary(srv))
+	}
+
+	fn(&sd.ListServicesOutput{
+		Services: services,
+	}, true)
+
+	return nil
+}
+
+func (s *AWSSDClientStub) RegisterInstance(input *sd.RegisterInstanceInput) (*sd.RegisterInstanceOutput, error) {
+
+	srvInstances, ok := s.instances[*input.ServiceId]
+	if !ok {
+		srvInstances = make(map[string]*sd.Instance)
+		s.instances[*input.ServiceId] = srvInstances
+	}
+
+	srvInstances[*input.InstanceId] = &sd.Instance{
+		Id:               input.InstanceId,
+		Attributes:       input.Attributes,
+		CreatorRequestId: input.CreatorRequestId,
+	}
+
+	return &sd.RegisterInstanceOutput{}, nil
+}
+
+func (s *AWSSDClientStub) UpdateService(input *sd.UpdateServiceInput) (*sd.UpdateServiceOutput, error) {
+	out, err := s.GetService(&sd.GetServiceInput{Id: input.Id})
+	if err != nil {
+		return nil, err
+	}
+
+	origSrv := out.Service
+	updateSrv := input.Service
+
+	origSrv.Description = updateSrv.Description
+	origSrv.DnsConfig.DnsRecords = updateSrv.DnsConfig.DnsRecords
+
+	return &sd.UpdateServiceOutput{}, nil
+}
+
+func newTestAWSSDProvider(api AWSSDClient, domainFilter DomainFilter, namespaceTypeFilter string) *AWSSDProvider {
+	return &AWSSDProvider{
+		client:              api,
+		namespaceFilter:     domainFilter,
+		namespaceTypeFilter: newSdNamespaceFilter(namespaceTypeFilter),
+		dryRun:              false,
+	}
+}
+
+func TestAWSSDProvider_Records(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": {
+			"a-srv": {
+				Id:          aws.String("a-srv"),
+				Name:        aws.String("service1"),
+				Description: aws.String("owner-id"),
+				DnsConfig: &sd.DnsConfig{
+					NamespaceId:   aws.String("private"),
+					RoutingPolicy: aws.String(sd.RoutingPolicyWeighted),
+					DnsRecords: []*sd.DnsRecord{{
+						Type: aws.String(sd.RecordTypeA),
+						TTL:  aws.Int64(100),
+					}},
+				},
+			},
+			"alias-srv": {
+				Id:          aws.String("alias-srv"),
+				Name:        aws.String("service2"),
+				Description: aws.String("owner-id"),
+				DnsConfig: &sd.DnsConfig{
+					NamespaceId:   aws.String("private"),
+					RoutingPolicy: aws.String(sd.RoutingPolicyWeighted),
+					DnsRecords: []*sd.DnsRecord{{
+						Type: aws.String(sd.RecordTypeA),
+						TTL:  aws.Int64(100),
+					}},
+				},
+			},
+			"cname-srv": {
+				Id:          aws.String("cname-srv"),
+				Name:        aws.String("service3"),
+				Description: aws.String("owner-id"),
+				DnsConfig: &sd.DnsConfig{
+					NamespaceId:   aws.String("private"),
+					RoutingPolicy: aws.String(sd.RoutingPolicyWeighted),
+					DnsRecords: []*sd.DnsRecord{{
+						Type: aws.String(sd.RecordTypeCname),
+						TTL:  aws.Int64(80),
+					}},
+				},
+			},
+		},
+	}
+
+	instances := map[string]map[string]*sd.Instance{
+		"a-srv": {
+			"1.2.3.4": {
+				Id: aws.String("1.2.3.4"),
+				Attributes: map[string]*string{
+					sdInstanceAttrIPV4: aws.String("1.2.3.4"),
+				},
+			},
+			"1.2.3.5": {
+				Id: aws.String("1.2.3.5"),
+				Attributes: map[string]*string{
+					sdInstanceAttrIPV4: aws.String("1.2.3.5"),
+				},
+			},
+		},
+		"alias-srv": {
+			"load-balancer.us-east-1.elb.amazonaws.com": {
+				Id: aws.String("load-balancer.us-east-1.elb.amazonaws.com"),
+				Attributes: map[string]*string{
+					sdInstanceAttrAlias: aws.String("load-balancer.us-east-1.elb.amazonaws.com"),
+				},
+			},
+		},
+		"cname-srv": {
+			"cname.target.com": {
+				Id: aws.String("cname.target.com"),
+				Attributes: map[string]*string{
+					sdInstanceAttrCname: aws.String("cname.target.com"),
+				},
+			},
+		},
+	}
+
+	expectedEndpoints := []*endpoint.Endpoint{
+		{DNSName: "service1.private.com", Targets: endpoint.Targets{"1.2.3.4", "1.2.3.5"}, RecordType: endpoint.RecordTypeA, RecordTTL: 100, Labels: map[string]string{endpoint.AWSSDDescriptionLabel: "owner-id"}},
+		{DNSName: "service2.private.com", Targets: endpoint.Targets{"load-balancer.us-east-1.elb.amazonaws.com"}, RecordType: endpoint.RecordTypeCNAME, RecordTTL: 100, Labels: map[string]string{endpoint.AWSSDDescriptionLabel: "owner-id"}},
+		{DNSName: "service3.private.com", Targets: endpoint.Targets{"cname.target.com"}, RecordType: endpoint.RecordTypeCNAME, RecordTTL: 80, Labels: map[string]string{endpoint.AWSSDDescriptionLabel: "owner-id"}},
+	}
+
+	api := &AWSSDClientStub{
+		namespaces: namespaces,
+		services:   services,
+		instances:  instances,
+	}
+
+	provider := newTestAWSSDProvider(api, NewDomainFilter([]string{}), "")
+
+	endpoints, _ := provider.Records()
+
+	assert.True(t, testutils.SameEndpoints(expectedEndpoints, endpoints), "expected and actual endpoints don't match, expected=%v, actual=%v", expectedEndpoints, endpoints)
+}
+
+func TestAWSSDProvider_ApplyChanges(t *testing.T) {
+	namespaces := map[string]*sd.Namespace{
+		"private": {
+			Id:   aws.String("private"),
+			Name: aws.String("private.com"),
+			Type: aws.String(sd.NamespaceTypeDnsPrivate),
+		},
+	}
+
+	api := &AWSSDClientStub{
+		namespaces: namespaces,
+		services:   make(map[string]map[string]*sd.Service),
+		instances:  make(map[string]map[string]*sd.Instance),
+	}
+
+	expectedEndpoints := []*endpoint.Endpoint{
+		{DNSName: "service1.private.com", Targets: endpoint.Targets{"1.2.3.4", "1.2.3.5"}, RecordType: endpoint.RecordTypeA, RecordTTL: 60},
+		{DNSName: "service2.private.com", Targets: endpoint.Targets{"load-balancer.us-east-1.elb.amazonaws.com"}, RecordType: endpoint.RecordTypeCNAME, RecordTTL: 80},
+		{DNSName: "service3.private.com", Targets: endpoint.Targets{"cname.target.com"}, RecordType: endpoint.RecordTypeCNAME, RecordTTL: 100},
+	}
+
+	provider := newTestAWSSDProvider(api, NewDomainFilter([]string{}), "")
+
+	// apply creates
+	provider.ApplyChanges(&plan.Changes{
+		Create: expectedEndpoints,
+	})
+
+	// make sure services were created
+	assert.Len(t, api.services["private"], 3)
+	existingServices, _ := provider.ListServicesByNamespaceID(namespaces["private"].Id)
+	assert.NotNil(t, existingServices["service1"])
+	assert.NotNil(t, existingServices["service2"])
+	assert.NotNil(t, existingServices["service3"])
+
+	// make sure instances were registered
+	endpoints, _ := provider.Records()
+	assert.True(t, testutils.SameEndpoints(expectedEndpoints, endpoints), "expected and actual endpoints don't match, expected=%v, actual=%v", expectedEndpoints, endpoints)
+
+	// apply deletes
+	provider.ApplyChanges(&plan.Changes{
+		Delete: expectedEndpoints,
+	})
+
+	// make sure all instances are gone
+	endpoints, _ = provider.Records()
+	assert.Empty(t, endpoints)
+}
+
+func TestAWSSDProvider_ListNamespaces(t *testing.T) {
+	namespaces := map[string]*sd.Namespace{
+		"private": {
+			Id:   aws.String("private"),
+			Name: aws.String("private.com"),
+			Type: aws.String(sd.NamespaceTypeDnsPrivate),
+		},
+		"public": {
+			Id:   aws.String("public"),
+			Name: aws.String("public.com"),
+			Type: aws.String(sd.NamespaceTypeDnsPublic),
+		},
+	}
+
+	api := &AWSSDClientStub{
+		namespaces: namespaces,
+	}
+
+	for _, tc := range []struct {
+		msg                 string
+		domainFilter        DomainFilter
+		namespaceTypeFilter string
+		expectedNamespaces  []*sd.NamespaceSummary
+	}{
+		{"public filter", NewDomainFilter([]string{}), "public", []*sd.NamespaceSummary{namespaceToNamespaceSummary(namespaces["public"])}},
+		{"private filter", NewDomainFilter([]string{}), "private", []*sd.NamespaceSummary{namespaceToNamespaceSummary(namespaces["private"])}},
+		{"domain filter", NewDomainFilter([]string{"public.com"}), "", []*sd.NamespaceSummary{namespaceToNamespaceSummary(namespaces["public"])}},
+		{"non-existing domain", NewDomainFilter([]string{"xxx.com"}), "", []*sd.NamespaceSummary{}},
+	} {
+		provider := newTestAWSSDProvider(api, tc.domainFilter, tc.namespaceTypeFilter)
+
+		result, err := provider.ListNamespaces()
+		require.NoError(t, err)
+
+		expectedMap := make(map[string]*sd.NamespaceSummary)
+		resultMap := make(map[string]*sd.NamespaceSummary)
+		for _, ns := range tc.expectedNamespaces {
+			expectedMap[*ns.Id] = ns
+		}
+		for _, ns := range result {
+			resultMap[*ns.Id] = ns
+		}
+
+		if !reflect.DeepEqual(resultMap, expectedMap) {
+			t.Errorf("AWSSDProvider.ListNamespaces() error = %v, wantErr %v", result, tc.expectedNamespaces)
+		}
+	}
+}
+
+func TestAWSSDProvider_ListServicesByNamespace(t *testing.T) {
+	namespaces := map[string]*sd.Namespace{
+		"private": {
+			Id:   aws.String("private"),
+			Name: aws.String("private.com"),
+			Type: aws.String(sd.NamespaceTypeDnsPrivate),
+		},
+		"public": {
+			Id:   aws.String("public"),
+			Name: aws.String("public.com"),
+			Type: aws.String(sd.NamespaceTypeDnsPublic),
+		},
+	}
+
+	services := map[string]map[string]*sd.Service{
+		"private": {
+			"srv1": {
+				Id:   aws.String("srv1"),
+				Name: aws.String("service1"),
+			},
+			"srv2": {
+				Id:   aws.String("srv2"),
+				Name: aws.String("service2"),
+			},
+		},
+		"public": {
+			"srv3": {
+				Id:   aws.String("srv3"),
+				Name: aws.String("service3"),
+			},
+		},
+	}
+
+	api := &AWSSDClientStub{
+		namespaces: namespaces,
+		services:   services,
+	}
+
+	for _, tc := range []struct {
+		expectedServices map[string]*sd.Service
+	}{
+		{map[string]*sd.Service{"service1": services["private"]["srv1"], "service2": services["private"]["srv2"]}},
+	} {
+		provider := newTestAWSSDProvider(api, NewDomainFilter([]string{}), "")
+
+		result, err := provider.ListServicesByNamespaceID(namespaces["private"].Id)
+		require.NoError(t, err)
+
+		if !reflect.DeepEqual(result, tc.expectedServices) {
+			t.Errorf("AWSSDProvider.ListServicesByNamespaceID() error = %v, wantErr %v", result, tc.expectedServices)
+		}
+	}
+}
+
+func TestAWSSDProvider_ListInstancesByService(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"),
+				Name: aws.String("service1"),
+			},
+			"srv2": {
+				Id:   aws.String("srv2"),
+				Name: aws.String("service2"),
+			},
+		},
+	}
+
+	instances := map[string]map[string]*sd.Instance{
+		"srv1": {
+			"inst1": {
+				Id: aws.String("inst1"),
+				Attributes: map[string]*string{
+					sdInstanceAttrIPV4: aws.String("1.2.3.4"),
+				},
+			},
+			"inst2": {
+				Id: aws.String("inst2"),
+				Attributes: map[string]*string{
+					sdInstanceAttrIPV4: aws.String("1.2.3.5"),
+				},
+			},
+		},
+	}
+
+	api := &AWSSDClientStub{
+		namespaces: namespaces,
+		services:   services,
+		instances:  instances,
+	}
+
+	provider := newTestAWSSDProvider(api, NewDomainFilter([]string{}), "")
+
+	result, err := provider.ListInstancesByServiceID(services["private"]["srv1"].Id)
+	require.NoError(t, err)
+
+	expectedInstances := []*sd.InstanceSummary{instanceToInstanceSummary(instances["srv1"]["inst1"]), instanceToInstanceSummary(instances["srv1"]["inst2"])}
+
+	expectedMap := make(map[string]*sd.InstanceSummary)
+	resultMap := make(map[string]*sd.InstanceSummary)
+	for _, inst := range expectedInstances {
+		expectedMap[*inst.Id] = inst
+	}
+	for _, inst := range result {
+		resultMap[*inst.Id] = inst
+	}
+
+	if !reflect.DeepEqual(resultMap, expectedMap) {
+		t.Errorf("AWSSDProvider.ListInstancesByServiceID() error = %v, wantErr %v", result, expectedInstances)
+	}
+}
+
+func TestAWSSDProvider_CreateService(t *testing.T) {
+	namespaces := map[string]*sd.Namespace{
+		"private": {
+			Id:   aws.String("private"),
+			Name: aws.String("private.com"),
+			Type: aws.String(sd.NamespaceTypeDnsPrivate),
+		},
+	}
+
+	api := &AWSSDClientStub{
+		namespaces: namespaces,
+		services:   make(map[string]map[string]*sd.Service),
+	}
+
+	expectedServices := make(map[string]*sd.Service)
+
+	provider := newTestAWSSDProvider(api, NewDomainFilter([]string{}), "")
+
+	// A type
+	provider.CreateService(aws.String("private"), aws.String("A-srv"), &endpoint.Endpoint{
+		RecordType: endpoint.RecordTypeA,
+		RecordTTL:  60,
+		Targets:    endpoint.Targets{"1.2.3.4"},
+	})
+	expectedServices["A-srv"] = &sd.Service{
+		Name: aws.String("A-srv"),
+		DnsConfig: &sd.DnsConfig{
+			NamespaceId:   aws.String("private"),
+			RoutingPolicy: aws.String(sd.RoutingPolicyMultivalue),
+			DnsRecords: []*sd.DnsRecord{{
+				Type: aws.String(sd.RecordTypeA),
+				TTL:  aws.Int64(60),
+			}},
+		},
+	}
+
+	// CNAME type
+	provider.CreateService(aws.String("private"), aws.String("CNAME-srv"), &endpoint.Endpoint{
+		RecordType: endpoint.RecordTypeCNAME,
+		RecordTTL:  80,
+		Targets:    endpoint.Targets{"cname.target.com"},
+	})
+	expectedServices["CNAME-srv"] = &sd.Service{
+		Name: aws.String("CNAME-srv"),
+		DnsConfig: &sd.DnsConfig{
+			NamespaceId:   aws.String("private"),
+			RoutingPolicy: aws.String(sd.RoutingPolicyWeighted),
+			DnsRecords: []*sd.DnsRecord{{
+				Type: aws.String(sd.RecordTypeCname),
+				TTL:  aws.Int64(80),
+			}},
+		},
+	}
+
+	// ALIAS type
+	provider.CreateService(aws.String("private"), aws.String("ALIAS-srv"), &endpoint.Endpoint{
+		RecordType: endpoint.RecordTypeCNAME,
+		RecordTTL:  100,
+		Targets:    endpoint.Targets{"load-balancer.us-east-1.elb.amazonaws.com"},
+	})
+	expectedServices["ALIAS-srv"] = &sd.Service{
+		Name: aws.String("ALIAS-srv"),
+		DnsConfig: &sd.DnsConfig{
+			NamespaceId:   aws.String("private"),
+			RoutingPolicy: aws.String(sd.RoutingPolicyWeighted),
+			DnsRecords: []*sd.DnsRecord{{
+				Type: aws.String(sd.RecordTypeA),
+				TTL:  aws.Int64(100),
+			}},
+		},
+	}
+
+	validateAWSSDServicesMapsEqual(t, expectedServices, api.services["private"])
+}
+
+func validateAWSSDServicesMapsEqual(t *testing.T, expected map[string]*sd.Service, services map[string]*sd.Service) {
+	require.Len(t, services, len(expected))
+
+	for _, srv := range services {
+		validateAWSSDServicesEqual(t, expected[*srv.Name], srv)
+	}
+}
+
+func validateAWSSDServicesEqual(t *testing.T, expected *sd.Service, srv *sd.Service) {
+	assert.Equal(t, aws.StringValue(expected.Description), aws.StringValue(srv.Description))
+	assert.Equal(t, aws.StringValue(expected.Name), aws.StringValue(srv.Name))
+	assert.True(t, reflect.DeepEqual(*expected.DnsConfig, *srv.DnsConfig))
+}
+
+func TestAWSSDProvider_UpdateService(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"),
+				Name: aws.String("service1"),
+				DnsConfig: &sd.DnsConfig{
+					NamespaceId:   aws.String("private"),
+					RoutingPolicy: aws.String(sd.RoutingPolicyMultivalue),
+					DnsRecords: []*sd.DnsRecord{{
+						Type: aws.String(sd.RecordTypeA),
+						TTL:  aws.Int64(60),
+					}},
+				},
+			},
+		},
+	}
+
+	api := &AWSSDClientStub{
+		namespaces: namespaces,
+		services:   services,
+	}
+
+	provider := newTestAWSSDProvider(api, NewDomainFilter([]string{}), "")
+
+	// update service with different TTL
+	provider.UpdateService(services["private"]["srv1"], &endpoint.Endpoint{
+		RecordType: endpoint.RecordTypeA,
+		RecordTTL:  100,
+	})
+
+	assert.Equal(t, int64(100), *api.services["private"]["srv1"].DnsConfig.DnsRecords[0].TTL)
+}
+
+func TestAWSSDProvider_RegisterInstance(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": {
+			"a-srv": {
+				Id:   aws.String("a-srv"),
+				Name: aws.String("service1"),
+				DnsConfig: &sd.DnsConfig{
+					NamespaceId:   aws.String("private"),
+					RoutingPolicy: aws.String(sd.RoutingPolicyWeighted),
+					DnsRecords: []*sd.DnsRecord{{
+						Type: aws.String(sd.RecordTypeA),
+						TTL:  aws.Int64(60),
+					}},
+				},
+			},
+			"cname-srv": {
+				Id:   aws.String("cname-srv"),
+				Name: aws.String("service2"),
+				DnsConfig: &sd.DnsConfig{
+					NamespaceId:   aws.String("private"),
+					RoutingPolicy: aws.String(sd.RoutingPolicyWeighted),
+					DnsRecords: []*sd.DnsRecord{{
+						Type: aws.String(sd.RecordTypeCname),
+						TTL:  aws.Int64(60),
+					}},
+				},
+			},
+			"alias-srv": {
+				Id:   aws.String("alias-srv"),
+				Name: aws.String("service3"),
+				DnsConfig: &sd.DnsConfig{
+					NamespaceId:   aws.String("private"),
+					RoutingPolicy: aws.String(sd.RoutingPolicyWeighted),
+					DnsRecords: []*sd.DnsRecord{{
+						Type: aws.String(sd.RecordTypeA),
+						TTL:  aws.Int64(60),
+					}},
+				},
+			},
+		},
+	}
+
+	api := &AWSSDClientStub{
+		namespaces: namespaces,
+		services:   services,
+		instances:  make(map[string]map[string]*sd.Instance),
+	}
+
+	provider := newTestAWSSDProvider(api, NewDomainFilter([]string{}), "")
+
+	expectedInstances := make(map[string]*sd.Instance)
+
+	// IP-based instance
+	provider.RegisterInstance(services["private"]["a-srv"], &endpoint.Endpoint{
+		RecordType: endpoint.RecordTypeA,
+		DNSName:    "service1.private.com.",
+		RecordTTL:  300,
+		Targets:    endpoint.Targets{"1.2.3.4", "1.2.3.5"},
+	})
+	expectedInstances["1.2.3.4"] = &sd.Instance{
+		Id: aws.String("1.2.3.4"),
+		Attributes: map[string]*string{
+			sdInstanceAttrIPV4: aws.String("1.2.3.4"),
+		},
+	}
+	expectedInstances["1.2.3.5"] = &sd.Instance{
+		Id: aws.String("1.2.3.5"),
+		Attributes: map[string]*string{
+			sdInstanceAttrIPV4: aws.String("1.2.3.5"),
+		},
+	}
+
+	// ALIAS instance
+	provider.RegisterInstance(services["private"]["alias-srv"], &endpoint.Endpoint{
+		RecordType: endpoint.RecordTypeCNAME,
+		DNSName:    "service1.private.com.",
+		RecordTTL:  300,
+		Targets:    endpoint.Targets{"load-balancer.us-east-1.elb.amazonaws.com", "load-balancer.us-west-2.elb.amazonaws.com"},
+	})
+	expectedInstances["load-balancer.us-east-1.elb.amazonaws.com"] = &sd.Instance{
+		Id: aws.String("load-balancer.us-east-1.elb.amazonaws.com"),
+		Attributes: map[string]*string{
+			sdInstanceAttrAlias: aws.String("load-balancer.us-east-1.elb.amazonaws.com"),
+		},
+	}
+	expectedInstances["load-balancer.us-west-2.elb.amazonaws.com"] = &sd.Instance{
+		Id: aws.String("load-balancer.us-west-2.elb.amazonaws.com"),
+		Attributes: map[string]*string{
+			sdInstanceAttrAlias: aws.String("load-balancer.us-west-2.elb.amazonaws.com"),
+		},
+	}
+
+	// CNAME instance
+	provider.RegisterInstance(services["private"]["cname-srv"], &endpoint.Endpoint{
+		RecordType: endpoint.RecordTypeCNAME,
+		DNSName:    "service2.private.com.",
+		RecordTTL:  300,
+		Targets:    endpoint.Targets{"cname.target.com"},
+	})
+	expectedInstances["cname.target.com"] = &sd.Instance{
+		Id: aws.String("cname.target.com"),
+		Attributes: map[string]*string{
+			sdInstanceAttrCname: aws.String("cname.target.com"),
+		},
+	}
+
+	// validate instances
+	for _, srvInst := range api.instances {
+		for id, inst := range srvInst {
+			if !reflect.DeepEqual(*expectedInstances[id], *inst) {
+				t.Errorf("Instances don't match, expected = %v, actual %v", *expectedInstances[id], *inst)
+			}
+		}
+	}
+}
+
+func TestAWSSDProvider_DeregisterInstance(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"),
+				Name: aws.String("service1"),
+			},
+		},
+	}
+
+	instances := map[string]map[string]*sd.Instance{
+		"srv1": {
+			"1.2.3.4": {
+				Id: aws.String("1.2.3.4"),
+				Attributes: map[string]*string{
+					sdInstanceAttrIPV4: aws.String("1.2.3.4"),
+				},
+			},
+		},
+	}
+
+	api := &AWSSDClientStub{
+		namespaces: namespaces,
+		services:   services,
+		instances:  instances,
+	}
+
+	provider := newTestAWSSDProvider(api, NewDomainFilter([]string{}), "")
+
+	provider.DeregisterInstance(services["private"]["srv1"], endpoint.NewEndpoint("srv1.private.com.", endpoint.RecordTypeA, "1.2.3.4"))
+
+	assert.Len(t, instances["srv1"], 0)
+}

registry/aws_sd_registry.go

@@ -0,0 +1,88 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package registry
+
+import (
+	"errors"
+
+	"github.com/kubernetes-incubator/external-dns/endpoint"
+	"github.com/kubernetes-incubator/external-dns/plan"
+	"github.com/kubernetes-incubator/external-dns/provider"
+)
+
+// AWSSDRegistry implements registry interface with ownership information associated via the Description field of SD Service
+type AWSSDRegistry struct {
+	provider provider.Provider
+	ownerID  string
+}
+
+// NewAWSSDRegistry returns implementation of registry for AWS SD
+func NewAWSSDRegistry(provider provider.Provider, ownerID string) (*AWSSDRegistry, error) {
+	if ownerID == "" {
+		return nil, errors.New("owner id cannot be empty")
+	}
+	return &AWSSDRegistry{
+		provider: provider,
+		ownerID:  ownerID,
+	}, nil
+}
+
+// Records calls AWS SD API and expects AWS SD provider to provider Owner/Resource information as a serialized
+// value in the AWSSDDescriptionLabel value in the Labels map
+func (sdr *AWSSDRegistry) Records() ([]*endpoint.Endpoint, error) {
+	records, err := sdr.provider.Records()
+	if err != nil {
+		return nil, err
+	}
+
+	for _, record := range records {
+		labels, err := endpoint.NewLabelsFromString(record.Labels[endpoint.AWSSDDescriptionLabel])
+		if err != nil {
+			// if we fail to parse the output then simply assume the endpoint is not managed by any instance of External DNS
+			record.Labels = endpoint.NewLabels()
+			continue
+		}
+		record.Labels = labels
+	}
+
+	return records, nil
+}
+
+// ApplyChanges filters out records not owned the External-DNS, additionally it adds the required label
+// inserted in the AWS SD instance as a CreateID field
+func (sdr *AWSSDRegistry) ApplyChanges(changes *plan.Changes) error {
+	filteredChanges := &plan.Changes{
+		Create:    changes.Create,
+		UpdateNew: filterOwnedRecords(sdr.ownerID, changes.UpdateNew),
+		UpdateOld: filterOwnedRecords(sdr.ownerID, changes.UpdateOld),
+		Delete:    filterOwnedRecords(sdr.ownerID, changes.Delete),
+	}
+
+	sdr.updateLabels(filteredChanges.Create)
+	sdr.updateLabels(filteredChanges.UpdateNew)
+	sdr.updateLabels(filteredChanges.UpdateOld)
+	sdr.updateLabels(filteredChanges.Delete)
+
+	return sdr.provider.ApplyChanges(filteredChanges)
+}
+
+func (sdr *AWSSDRegistry) updateLabels(endpoints []*endpoint.Endpoint) {
+	for _, ep := range endpoints {
+		ep.Labels[endpoint.OwnerLabelKey] = sdr.ownerID
+		ep.Labels[endpoint.AWSSDDescriptionLabel] = ep.Labels.Serialize(false)
+	}
+}

registry/aws_sd_registry_test.go

@@ -0,0 +1,163 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package registry
+
+import (
+	"testing"
+
+	"github.com/kubernetes-incubator/external-dns/endpoint"
+	"github.com/kubernetes-incubator/external-dns/internal/testutils"
+	"github.com/kubernetes-incubator/external-dns/plan"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type inMemoryProvider struct {
+	endpoints      []*endpoint.Endpoint
+	onApplyChanges func(changes *plan.Changes)
+}
+
+func (p *inMemoryProvider) Records() ([]*endpoint.Endpoint, error) {
+	return p.endpoints, nil
+}
+
+func (p *inMemoryProvider) ApplyChanges(changes *plan.Changes) error {
+	p.onApplyChanges(changes)
+	return nil
+}
+
+func newInMemoryProvider(endpoints []*endpoint.Endpoint, onApplyChanges func(changes *plan.Changes)) *inMemoryProvider {
+	return &inMemoryProvider{
+		endpoints:      endpoints,
+		onApplyChanges: onApplyChanges,
+	}
+}
+
+func TestAWSSDRegistry_NewAWSSDRegistry(t *testing.T) {
+	p := newInMemoryProvider(nil, nil)
+	_, err := NewAWSSDRegistry(p, "")
+	require.Error(t, err)
+
+	_, err = NewAWSSDRegistry(p, "owner")
+	require.NoError(t, err)
+}
+
+func TestAWSSDRegistryTest_Records(t *testing.T) {
+	p := newInMemoryProvider([]*endpoint.Endpoint{
+		newEndpointWithOwnerAndDescription("foo1.test-zone.example.org", "1.2.3.4", endpoint.RecordTypeA, "", ""),
+		newEndpointWithOwnerAndDescription("foo2.test-zone.example.org", "1.2.3.4", endpoint.RecordTypeA, "owner", "\"heritage=external-dns,external-dns/owner=owner\""),
+		newEndpointWithOwnerAndDescription("foo3.test-zone.example.org", "my-domain.com", endpoint.RecordTypeCNAME, "", ""),
+		newEndpointWithOwnerAndDescription("foo4.test-zone.example.org", "my-domain.com", endpoint.RecordTypeCNAME, "owner", "\"heritage=external-dns,external-dns/owner=owner\""),
+	}, nil)
+	expectedRecords := []*endpoint.Endpoint{
+		{
+			DNSName:    "foo1.test-zone.example.org",
+			Targets:    endpoint.Targets{"1.2.3.4"},
+			RecordType: endpoint.RecordTypeA,
+			Labels: map[string]string{
+				endpoint.OwnerLabelKey: "",
+			},
+		},
+		{
+			DNSName:    "foo2.test-zone.example.org",
+			Targets:    endpoint.Targets{"1.2.3.4"},
+			RecordType: endpoint.RecordTypeA,
+			Labels: map[string]string{
+				endpoint.OwnerLabelKey: "owner",
+			},
+		},
+		{
+			DNSName:    "foo3.test-zone.example.org",
+			Targets:    endpoint.Targets{"my-domain.com"},
+			RecordType: endpoint.RecordTypeCNAME,
+			Labels: map[string]string{
+				endpoint.OwnerLabelKey: "",
+			},
+		},
+		{
+			DNSName:    "foo4.test-zone.example.org",
+			Targets:    endpoint.Targets{"my-domain.com"},
+			RecordType: endpoint.RecordTypeCNAME,
+			Labels: map[string]string{
+				endpoint.OwnerLabelKey: "owner",
+			},
+		},
+	}
+
+	r, _ := NewAWSSDRegistry(p, "owner")
+	records, _ := r.Records()
+
+	assert.True(t, testutils.SameEndpoints(records, expectedRecords))
+}
+
+func TestAWSSDRegistry_Records_ApplyChanges(t *testing.T) {
+	changes := &plan.Changes{
+		Create: []*endpoint.Endpoint{
+			newEndpointWithOwner("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", endpoint.RecordTypeCNAME, "owner"),
+		},
+		Delete: []*endpoint.Endpoint{
+			newEndpointWithOwner("foobar.test-zone.example.org", "1.2.3.4", endpoint.RecordTypeA, "owner"),
+		},
+		UpdateNew: []*endpoint.Endpoint{
+			newEndpointWithOwner("tar.test-zone.example.org", "new-tar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner"),
+		},
+		UpdateOld: []*endpoint.Endpoint{
+			newEndpointWithOwner("tar.test-zone.example.org", "tar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner"),
+		},
+	}
+	expected := &plan.Changes{
+		Create: []*endpoint.Endpoint{
+			newEndpointWithOwnerAndDescription("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", endpoint.RecordTypeCNAME, "owner", "\"heritage=external-dns,external-dns/owner=owner\""),
+		},
+		Delete: []*endpoint.Endpoint{
+			newEndpointWithOwnerAndDescription("foobar.test-zone.example.org", "1.2.3.4", endpoint.RecordTypeA, "owner", "\"heritage=external-dns,external-dns/owner=owner\""),
+		},
+		UpdateNew: []*endpoint.Endpoint{
+			newEndpointWithOwnerAndDescription("tar.test-zone.example.org", "new-tar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner", "\"heritage=external-dns,external-dns/owner=owner\""),
+		},
+		UpdateOld: []*endpoint.Endpoint{
+			newEndpointWithOwnerAndDescription("tar.test-zone.example.org", "tar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner", "\"heritage=external-dns,external-dns/owner=owner\""),
+		},
+	}
+	p := newInMemoryProvider(nil, func(got *plan.Changes) {
+		mExpected := map[string][]*endpoint.Endpoint{
+			"Create":    expected.Create,
+			"UpdateNew": expected.UpdateNew,
+			"UpdateOld": expected.UpdateOld,
+			"Delete":    expected.Delete,
+		}
+		mGot := map[string][]*endpoint.Endpoint{
+			"Create":    got.Create,
+			"UpdateNew": got.UpdateNew,
+			"UpdateOld": got.UpdateOld,
+			"Delete":    got.Delete,
+		}
+		assert.True(t, testutils.SamePlanChanges(mGot, mExpected))
+	})
+	r, err := NewAWSSDRegistry(p, "owner")
+	require.NoError(t, err)
+
+	err = r.ApplyChanges(changes)
+	require.NoError(t, err)
+}
+
+func newEndpointWithOwnerAndDescription(dnsName, target, recordType, ownerID string, description string) *endpoint.Endpoint {
+	e := endpoint.NewEndpoint(dnsName, recordType, target)
+	e.Labels[endpoint.OwnerLabelKey] = ownerID
+	e.Labels[endpoint.AWSSDDescriptionLabel] = description
+	return e
+}