func (a *awsConsumer) getAssignedTXTRecordObject(aliasRecord *route53.ResourceRecordSet) *route53.ResourceRecordSet {
	return &route53.ResourceRecordSet{
		Type: aws.String("TXT"),
		Name: aliasRecord.Name,
		TTL:  aws.Int64(defaultTxtTTL),
		ResourceRecords: []*route53.ResourceRecord{{
			Value: aws.String(a.getGroupID()),
		}},
	}
}
getAssignedTXTRecordObject returns the TXT record which accompanies the Alias record
package consumers
import (
	"errors"
	"fmt"
	"strings"
	"sync"
	log "github.com/Sirupsen/logrus"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/route53"
	"github.com/zalando-incubator/mate/pkg"
	awsclient "github.com/zalando-incubator/mate/pkg/aws"
)
type AWSClient interface {
	ListRecordSets(zoneID string) ([]*route53.ResourceRecordSet, error)
	ChangeRecordSets(upsert, del, create []*route53.ResourceRecordSet, zoneID string) error
	GetCanonicalZoneIDs(lbDNS []string) (map[string]string, error)  
	GetHostedZones() (map[string]string, error)                     
}
type awsConsumer struct {
	groupID string
	client  AWSClient
}
const (
	evaluateTargetHealth = true
	defaultTxtTTL        = int64(300)
	defaultATTL          = int64(300)
)
func NewAWSRoute53Consumer(awsRecordGroupID string) (Consumer, error) {
	if awsRecordGroupID == "" {
		return nil, errors.New("please provide --aws-record-group-id")
	}
	return withClient(awsclient.New(awsclient.Options{}), awsRecordGroupID), nil
}
func withClient(c AWSClient, groupID string) *awsConsumer {
	return &awsConsumer{
		groupID: groupID,
		client:  c,
	}
}
func (a *awsConsumer) Sync(endpoints []*pkg.Endpoint) error {
	kubeRecords, err := a.endpointsToRecords(endpoints)
	if err != nil {
		log.Errorf("failed to convert endpoints to RRS: %v. Aborting sync...", err)
		return err
	}
	hostedZonesMap, err := a.client.GetHostedZones()
	if err != nil {
		return err
	}
	if len(hostedZonesMap) == 0 {
		log.Warnln("No hosted zones found in Route53. At least one hosted zone should be created to create DNS records...")
		return nil
	}
	inputByZoneID := map[string][]*route53.ResourceRecordSet{}
	for _, record := range kubeRecords {
		zoneID := getZoneIDForEndpoint(hostedZonesMap, record)  
		if zoneID == "" {
			log.Warnf("Hosted zone for endpoint: %s was not found. Skipping record...", aws.StringValue(record.Name))
			continue
		}
		inputByZoneID[zoneID] = append(inputByZoneID[zoneID], record)
	}
	var wg sync.WaitGroup
	for zoneName, zoneID := range hostedZonesMap {
		wg.Add(1)
		go func(zoneName, zoneID string) {
			defer wg.Done()
			err := a.syncPerHostedZone(inputByZoneID[zoneID], zoneID)
			if err != nil {
				log.Errorf("Error changing records per zone: %s. Error: %v", zoneName, err)
			}
		}(zoneName, zoneID)
	}
	wg.Wait()
	return nil
}
func (a *awsConsumer) syncPerHostedZone(kubeRecords []*route53.ResourceRecordSet, zoneID string) error {
	existingRecords, err := a.client.ListRecordSets(zoneID)
	if err != nil {
		log.Errorf("failed to list records in zoneID: %s. Error: %v", zoneID, err)
		return err
	}
	recordInfoMap := a.recordInfo(existingRecords)
	var upsert, del []*route53.ResourceRecordSet
	upsertedMap := make(map[string]bool)  
	targetMap := map[string][]*string{}   
	for _, kr := range kubeRecords {
		targetMap[aws.StringValue(kr.Name)] = append(targetMap[aws.StringValue(kr.Name)], aws.String(a.getRecordTarget(kr)))
	}
	for _, kubeRecord := range kubeRecords {
		if _, upserted := upsertedMap[aws.StringValue(kubeRecord.Name)]; upserted {
			continue
		}
		existingRecordInfo, exist := recordInfoMap[aws.StringValue(kubeRecord.Name)]
		if !exist {  
			newTXTRecord := a.getAssignedTXTRecordObject(kubeRecord)
			upsert = append(upsert, kubeRecord, newTXTRecord)
			upsertedMap[aws.StringValue(kubeRecord.Name)] = true
			continue
		}
		if existingRecordInfo.GroupID != a.getGroupID() {  
			log.Warnf("Skipping record %s: with a group ID: %s", aws.StringValue(kubeRecord.Name), existingRecordInfo.GroupID)
			continue
		}
		kubeTargetsForDNS := targetMap[aws.StringValue(kubeRecord.Name)]
		targetStillRequired := false
		for _, targetPtr := range kubeTargetsForDNS {
			if pkg.SameDNSName(aws.StringValue(targetPtr), existingRecordInfo.Target) {
				targetStillRequired = true
				break
			}
		}
		if !targetStillRequired {  
			newTXTRecord := a.getAssignedTXTRecordObject(kubeRecord)
			upsert = append(upsert, kubeRecord, newTXTRecord)
			upsertedMap[aws.StringValue(kubeRecord.Name)] = true
		}
	}
	for _, existingRecord := range existingRecords {
		recordInfo := recordInfoMap[aws.StringValue(existingRecord.Name)]
		if recordInfo.GroupID == a.getGroupID() {
			remove := true
			for _, kubeRecord := range kubeRecords {
				if pkg.SameDNSName(aws.StringValue(kubeRecord.Name), aws.StringValue(existingRecord.Name)) {
					remove = false
				}
			}
			if remove {
				del = append(del, existingRecord)
			}
		}
	}
	if len(upsert) > 0 || len(del) > 0 {
		log.Debugln("Records to be upserted: ", upsert)
		log.Debugln("Records to be deleted: ", del)
		return a.client.ChangeRecordSets(upsert, del, nil, zoneID)
	}
	log.Infoln("No changes submitted for zone: ", zoneID)
	return nil
}
func (a *awsConsumer) Consume(endpoints <-chan *pkg.Endpoint, errors chan<- error, done <-chan struct{}, wg *sync.WaitGroup) {
	wg.Add(1)
	defer wg.Done()
	log.Infoln("[AWS] Listening for events...")
	for {
		select {
		case e, ok := <-endpoints:
			if !ok {
				log.Info("[AWS] channel closed")
				return
			}
			log.Infof("[AWS] Processing (%s, %s, %s)\n", e.DNSName, e.IP, e.Hostname)
			err := a.Process(e)
			if err != nil {
				errors <- err
			}
		case <-done:
			log.Info("[AWS] Exited consuming loop.")
			return
		}
	}
}
func (a *awsConsumer) Process(endpoint *pkg.Endpoint) error {
	hostedZonesMap, err := a.client.GetHostedZones()
	if err != nil {
		return err
	}
	ARecords, err := a.endpointsToRecords([]*pkg.Endpoint{endpoint})
	if err != nil {
		log.Errorf("failed to convert endpoint to RRS: %v. Aborting process...", err)
		return err
	}
	if len(ARecords) != 1 {
		return fmt.Errorf("failed to process endpoint. A record could not be constructed for: %s:%s:%s", endpoint.DNSName, endpoint.Hostname, endpoint.IP)
	}
	create := []*route53.ResourceRecordSet{ARecords[0], a.getAssignedTXTRecordObject(ARecords[0])}
	zoneID := getZoneIDForEndpoint(hostedZonesMap, ARecords[0])
	if zoneID == "" {
		log.Warnf("Hosted zone for endpoint: %s was not found. Skipping record...", endpoint.DNSName)
		return nil
	}
	err = a.client.ChangeRecordSets(nil, nil, create, zoneID)
	if err != nil && strings.Contains(err.Error(), "already exists") {
		log.Warnf("Record [name=%s] could not be created, another record with same name already exists", endpoint.DNSName)
		return nil
	}
	return err
}
func getZoneIDForEndpoint(hostedZonesMap map[string]string, record *route53.ResourceRecordSet) string {
	var matchName string
	var matchID string
	for zoneName, zoneID := range hostedZonesMap {
		if strings.HasSuffix(aws.StringValue(record.Name), zoneName) && len(zoneName) > len(matchName) {  
			matchName = zoneName
			matchID = zoneID
		}
	}
	return matchID
}
func (a *awsConsumer) getGroupID() string {
	return fmt.Sprintf("\"mate:%s\"", a.groupID)
}