mirrors/external-dns
1
0
Fork 0
mirror of https://github.com/kubernetes-sigs/external-dns.git synced 2025-08-05 09:06:58 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #4640 from mjlshen/aws-sdk-go-v2

Browse Source 
Migrate to aws-sdk-go-v2
This commit is contained in:
Kubernetes Prow Robot 2024-09-07 08:55:19 +01:00 committed by GitHub
commit 4333b31db2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 1249 additions and 1156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
go.mod
View File

@ -16,7 +16,14 @@ require (
github.com/akamai/AkamaiOPEN-edgegrid-golang v1.2.2
github.com/alecthomas/kingpin/v2 v2.4.0
github.com/aliyun/alibaba-cloud-sdk-go v1.63.0
github.com/aws/aws-sdk-go v1.55.5
github.com/aws/aws-sdk-go-v2 v1.30.3
github.com/aws/aws-sdk-go-v2/config v1.27.27
github.com/aws/aws-sdk-go-v2/credentials v1.17.27
github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue v1.14.10
github.com/aws/aws-sdk-go-v2/service/dynamodb v1.34.4
github.com/aws/aws-sdk-go-v2/service/route53 v1.42.3
github.com/aws/aws-sdk-go-v2/service/servicediscovery v1.31.3
github.com/aws/aws-sdk-go-v2/service/sts v1.30.3
github.com/bodgit/tsig v1.2.2
github.com/cenkalti/backoff/v4 v4.3.0
github.com/civo/civogo v0.3.73
@ -85,6 +92,17 @@ require (
github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137 // indirect
github.com/alexbrainman/sspi v0.0.0-20180613141037-e580b900e9f5 // indirect
github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.11 // indirect
github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.15 // indirect
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.15 // indirect
github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0 // indirect
github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.22.3 // indirect
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.3 // indirect
github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.9.16 // indirect
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.17 // indirect
github.com/aws/aws-sdk-go-v2/service/sso v1.22.4 // indirect
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.26.4 // indirect
github.com/aws/smithy-go v1.20.3 // indirect
github.com/benbjohnson/clock v1.3.0 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect

40
go.sum
View File

@ -116,9 +116,45 @@ github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2/go.mod h1:W
github.com/aws/aws-lambda-go v1.13.3/go.mod h1:4UKl9IzQMoD+QF79YdCuzCwp8VbmG4VAQwij/eHl5CU=
github.com/aws/aws-sdk-go v1.15.11/go.mod h1:mFuSZ37Z9YOHbQEwBWztmVzqXrEkub65tZoCYDt7FT0=
github.com/aws/aws-sdk-go v1.27.0/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
github.com/aws/aws-sdk-go v1.55.5 h1:KKUZBfBoyqy5d3swXyiC7Q76ic40rYcbqH7qjh59kzU=
github.com/aws/aws-sdk-go v1.55.5/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
github.com/aws/aws-sdk-go-v2 v0.18.0/go.mod h1:JWVYvqSMppoMJC0x5wdwiImzgXTI9FuZwxzkQq9wy+g=
github.com/aws/aws-sdk-go-v2 v1.30.3 h1:jUeBtG0Ih+ZIFH0F4UkmL9w3cSpaMv9tYYDbzILP8dY=
github.com/aws/aws-sdk-go-v2 v1.30.3/go.mod h1:nIQjQVp5sfpQcTc9mPSr1B0PaWK5ByX9MOoDadSN4lc=
github.com/aws/aws-sdk-go-v2/config v1.27.27 h1:HdqgGt1OAP0HkEDDShEl0oSYa9ZZBSOmKpdpsDMdO90=
github.com/aws/aws-sdk-go-v2/config v1.27.27/go.mod h1:MVYamCg76dFNINkZFu4n4RjDixhVr51HLj4ErWzrVwg=
github.com/aws/aws-sdk-go-v2/credentials v1.17.27 h1:2raNba6gr2IfA0eqqiP2XiQ0UVOpGPgDSi0I9iAP+UI=
github.com/aws/aws-sdk-go-v2/credentials v1.17.27/go.mod h1:gniiwbGahQByxan6YjQUMcW4Aov6bLC3m+evgcoN4r4=
github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue v1.14.10 h1:orAIBscNu5aIjDOnKIrjO+IUFPMLKj3Lp0bPf4chiPc=
github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue v1.14.10/go.mod h1:GNjJ8daGhv10hmQYCnmkV8HuY6xXOXV4vzBssSjEIlU=
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.11 h1:KreluoV8FZDEtI6Co2xuNk/UqI9iwMrOx/87PBNIKqw=
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.11/go.mod h1:SeSUYBLsMYFoRvHE0Tjvn7kbxaUhl75CJi1sbfhMxkU=
github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.15 h1:SoNJ4RlFEQEbtDcCEt+QG56MY4fm4W8rYirAmq+/DdU=
github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.15/go.mod h1:U9ke74k1n2bf+RIgoX1SXFed1HLs51OgUSs+Ph0KJP8=
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.15 h1:C6WHdGnTDIYETAm5iErQUiVNsclNx9qbJVPIt03B6bI=
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.15/go.mod h1:ZQLZqhcu+JhSrA9/NXRm8SkDvsycE+JkV3WGY41e+IM=
github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0 h1:hT8rVHwugYE2lEfdFE0QWVo81lF7jMrYJVDWI+f+VxU=
github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0/go.mod h1:8tu/lYfQfFe6IGnaOdrpVgEL2IrrDOf6/m9RQum4NkY=
github.com/aws/aws-sdk-go-v2/service/dynamodb v1.34.4 h1:utG3S4T+X7nONPIpRoi1tVcQdAdJxntiVS2yolPJyXc=
github.com/aws/aws-sdk-go-v2/service/dynamodb v1.34.4/go.mod h1:q9vzW3Xr1KEXa8n4waHiFt1PrppNDlMymlYP+xpsFbY=
github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.22.3 h1:r27/FnxLPixKBRIlslsvhqscBuMK8uysCYG9Kfgm098=
github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.22.3/go.mod h1:jqOFyN+QSWSoQC+ppyc4weiO8iNQXbzRbxDjQ1ayYd4=
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.3 h1:dT3MqvGhSoaIhRseqw2I0yH81l7wiR2vjs57O51EAm8=
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.3/go.mod h1:GlAeCkHwugxdHaueRr4nhPuY+WW+gR8UjlcqzPr1SPI=
github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.9.16 h1:lhAX5f7KpgwyieXjbDnRTjPEUI0l3emSRyxXj1PXP8w=
github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.9.16/go.mod h1:AblAlCwvi7Q/SFowvckgN+8M3uFPlopSYeLlbNDArhA=
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.17 h1:HGErhhrxZlQ044RiM+WdoZxp0p+EGM62y3L6pwA4olE=
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.17/go.mod h1:RkZEx4l0EHYDJpWppMJ3nD9wZJAa8/0lq9aVC+r2UII=
github.com/aws/aws-sdk-go-v2/service/route53 v1.42.3 h1:MmLCRqP4U4Cw9gJ4bNrCG0mWqEtBlmAVleyelcHARMU=
github.com/aws/aws-sdk-go-v2/service/route53 v1.42.3/go.mod h1:AMPjK2YnRh0YgOID3PqhJA1BRNfXDfGOnSsKHtAe8yA=
github.com/aws/aws-sdk-go-v2/service/servicediscovery v1.31.3 h1:EthA93BNgTnk36FoI9DCKtv4S0m63WzdGDYlBp/CvHQ=
github.com/aws/aws-sdk-go-v2/service/servicediscovery v1.31.3/go.mod h1:4xh/h0pevPhBkA4b2iYosZaqrThccxFREQxiGuZpJlc=
github.com/aws/aws-sdk-go-v2/service/sso v1.22.4 h1:BXx0ZIxvrJdSgSvKTZ+yRBeSqqgPM89VPlulEcl37tM=
github.com/aws/aws-sdk-go-v2/service/sso v1.22.4/go.mod h1:ooyCOXjvJEsUw7x+ZDHeISPMhtwI3ZCB7ggFMcFfWLU=
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.26.4 h1:yiwVzJW2ZxZTurVbYWA7QOrAaCYQR72t0wrSBfoesUE=
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.26.4/go.mod h1:0oxfLkpz3rQ/CHlx5hB7H69YUpFiI1tql6Q6Ne+1bCw=
github.com/aws/aws-sdk-go-v2/service/sts v1.30.3 h1:ZsDKRLXGWHk8WdtyYMoGNO7bTudrvuKpDKgMVRlepGE=
github.com/aws/aws-sdk-go-v2/service/sts v1.30.3/go.mod h1:zwySh8fpFyXp9yOr/KVzxOl8SRqgf/IDw5aUt9UKFcQ=
github.com/aws/smithy-go v1.20.3 h1:ryHwveWzPV5BIof6fyDvor6V3iUL7nTfiTKXHiW05nE=
github.com/aws/smithy-go v1.20.3/go.mod h1:krry+ya/rV9RDcV/Q16kpu6ypI4K2czasz0NC3qS14E=
github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A=
github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
github.com/beorn7/perks v0.0.0-20160804104726-4c0e84591b9a/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=

27
main.go
View File

@ -25,10 +25,9 @@ import (
"syscall"
"time"
awsSDK "github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/dynamodb"
"github.com/aws/aws-sdk-go/service/route53"
sd "github.com/aws/aws-sdk-go/service/servicediscovery"
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
"github.com/aws/aws-sdk-go-v2/service/route53"
sd "github.com/aws/aws-sdk-go-v2/service/servicediscovery"
"github.com/go-logr/logr"
"github.com/prometheus/client_golang/prometheus/promhttp"
log "github.com/sirupsen/logrus"
@ -206,10 +205,10 @@ func main() {
case "alibabacloud":
p, err = alibabacloud.NewAlibabaCloudProvider(cfg.AlibabaCloudConfigFile, domainFilter, zoneIDFilter, cfg.AlibabaCloudZoneType, cfg.DryRun)
case "aws":
sessions := aws.CreateSessions(cfg)
clients := make(map[string]aws.Route53API, len(sessions))
for profile, session := range sessions {
clients[profile] = route53.New(session)
configs := aws.CreateV2Configs(cfg)
clients := make(map[string]aws.Route53API, len(configs))
for profile, config := range configs {
clients[profile] = route53.NewFromConfig(config)
}
p, err = aws.NewAWSProvider(
@ -236,7 +235,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.DryRun, cfg.AWSSDServiceCleanup, cfg.TXTOwnerID, sd.New(aws.CreateDefaultSession(cfg)))
p, err = awssd.NewAWSSDProvider(domainFilter, cfg.AWSZoneType, cfg.DryRun, cfg.AWSSDServiceCleanup, cfg.TXTOwnerID, sd.NewFromConfig(aws.CreateDefaultV2Config(cfg)))
case "azure-dns", "azure":
p, err = azure.NewAzureProvider(cfg.AzureConfigFile, domainFilter, zoneNameFilter, zoneIDFilter, cfg.AzureSubscriptionID, cfg.AzureResourceGroup, cfg.AzureUserAssignedIdentityClientID, cfg.AzureActiveDirectoryAuthorityHost, cfg.DryRun)
case "azure-private-dns":
@ -383,11 +382,15 @@ func main() {
var r registry.Registry
switch cfg.Registry {
case "dynamodb":
config := awsSDK.NewConfig()
var dynamodbOpts []func(*dynamodb.Options)
if cfg.AWSDynamoDBRegion != "" {
config = config.WithRegion(cfg.AWSDynamoDBRegion)
dynamodbOpts = []func(*dynamodb.Options){
func(opts *dynamodb.Options) {
opts.Region = cfg.AWSDynamoDBRegion
},
}
}
r, err = registry.NewDynamoDBRegistry(p, cfg.TXTOwnerID, dynamodb.New(aws.CreateDefaultSession(cfg), config), cfg.AWSDynamoDBTable, cfg.TXTPrefix, cfg.TXTSuffix, cfg.TXTWildcardReplacement, cfg.ManagedDNSRecordTypes, cfg.ExcludeDNSRecordTypes, []byte(cfg.TXTEncryptAESKey), cfg.TXTCacheInterval)
r, err = registry.NewDynamoDBRegistry(p, cfg.TXTOwnerID, dynamodb.NewFromConfig(aws.CreateDefaultV2Config(cfg), dynamodbOpts...), cfg.AWSDynamoDBTable, cfg.TXTPrefix, cfg.TXTSuffix, cfg.TXTWildcardReplacement, cfg.ManagedDNSRecordTypes, cfg.ExcludeDNSRecordTypes, []byte(cfg.TXTEncryptAESKey), cfg.TXTCacheInterval)
case "noop":
r, err = registry.NewNoopRegistry(p)
case "txt":

408
provider/aws/aws.go
View File

@ -25,10 +25,9 @@ import (
"strings"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/awserr"
"github.com/aws/aws-sdk-go/aws/request"
"github.com/aws/aws-sdk-go/service/route53"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/route53"
route53types "github.com/aws/aws-sdk-go-v2/service/route53/types"
log "github.com/sirupsen/logrus"
"sigs.k8s.io/external-dns/endpoint"
@ -41,11 +40,11 @@ const (
recordTTL = 300
// From the experiments, it seems that the default MaxItems applied is 100,
// and that, on the server side, there is a hard limit of 300 elements per page.
// After a discussion with AWS representants, clients should accept
// when less items are returned, and still paginate accordingly.
// After a discussion with AWS representatives, clients should accept
// when fewer items are returned, and still paginate accordingly.
// As we are using the standard AWS client, this should already be compliant.
// Hence, ifever AWS decides to raise this limit, we will automatically reduce the pressure on rate limits
route53PageSize = "300"
// Hence, if AWS ever decides to raise this limit, we will automatically reduce the pressure on rate limits
route53PageSize int32 = 300
// providerSpecificAlias specifies whether a CNAME endpoint maps to an AWS ALIAS record.
providerSpecificAlias = "alias"
providerSpecificTargetHostedZone = "aws/target-hosted-zone"
@ -199,16 +198,16 @@ var canonicalHostedZones = map[string]string{
// Route53API is the subset of the AWS Route53 API that we actually use. Add methods as required. Signatures must match exactly.
// mostly taken from: https://github.com/kubernetes/kubernetes/blob/853167624edb6bc0cfdcdfb88e746e178f5db36c/federation/pkg/dnsprovider/providers/aws/route53/stubs/route53api.go
type Route53API interface {
ListResourceRecordSetsPagesWithContext(ctx context.Context, input *route53.ListResourceRecordSetsInput, fn func(resp *route53.ListResourceRecordSetsOutput, lastPage bool) (shouldContinue bool), opts ...request.Option) error
ChangeResourceRecordSetsWithContext(ctx context.Context, input *route53.ChangeResourceRecordSetsInput, opts ...request.Option) (*route53.ChangeResourceRecordSetsOutput, error)
CreateHostedZoneWithContext(ctx context.Context, input *route53.CreateHostedZoneInput, opts ...request.Option) (*route53.CreateHostedZoneOutput, error)
ListHostedZonesPagesWithContext(ctx context.Context, input *route53.ListHostedZonesInput, fn func(resp *route53.ListHostedZonesOutput, lastPage bool) (shouldContinue bool), opts ...request.Option) error
ListTagsForResourceWithContext(ctx context.Context, input *route53.ListTagsForResourceInput, opts ...request.Option) (*route53.ListTagsForResourceOutput, error)
ListResourceRecordSets(ctx context.Context, input *route53.ListResourceRecordSetsInput, optFns ...func(options *route53.Options)) (*route53.ListResourceRecordSetsOutput, error)
ChangeResourceRecordSets(ctx context.Context, input *route53.ChangeResourceRecordSetsInput, optFns ...func(options *route53.Options)) (*route53.ChangeResourceRecordSetsOutput, error)
CreateHostedZone(ctx context.Context, input *route53.CreateHostedZoneInput, optFns ...func(*route53.Options)) (*route53.CreateHostedZoneOutput, error)
ListHostedZones(ctx context.Context, input *route53.ListHostedZonesInput, optFns ...func(options *route53.Options)) (*route53.ListHostedZonesOutput, error)
ListTagsForResource(ctx context.Context, input *route53.ListTagsForResourceInput, optFns ...func(options *route53.Options)) (*route53.ListTagsForResourceOutput, error)
}
// wrapper to handle ownership relation throughout the provider implementation
type Route53Change struct {
route53.Change
route53types.Change
OwnedRecord string
sizeBytes int
sizeValues int
@ -218,13 +217,13 @@ type Route53Changes []*Route53Change
type profiledZone struct {
profile string
zone *route53.HostedZone
zone *route53types.HostedZone
}
func (cs Route53Changes) Route53Changes() []*route53.Change {
ret := []*route53.Change{}
func (cs Route53Changes) Route53Changes() []route53types.Change {
ret := []route53types.Change{}
for _, c := range cs {
ret = append(ret, &c.Change)
ret = append(ret, c.Change)
}
return ret
}
@ -253,7 +252,7 @@ type AWSProvider struct {
zoneTypeFilter provider.ZoneTypeFilter
// filter hosted zones by tags
zoneTagFilter provider.ZoneTagFilter
// extend filter for sub-domains in the zone (e.g. first.us-east-1.example.com)
// extend filter for subdomains in the zone (e.g. first.us-east-1.example.com)
zoneMatchParent bool
preferCNAME bool
zonesCache *zonesListCache
@ -302,13 +301,13 @@ func NewAWSProvider(awsConfig AWSConfig, clients map[string]Route53API) (*AWSPro
}
// Zones returns the list of hosted zones.
func (p *AWSProvider) Zones(ctx context.Context) (map[string]*route53.HostedZone, error) {
func (p *AWSProvider) Zones(ctx context.Context) (map[string]*route53types.HostedZone, error) {
zones, err := p.zones(ctx)
if err != nil {
return nil, err
}
result := make(map[string]*route53.HostedZone, len(zones))
result := make(map[string]*route53types.HostedZone, len(zones))
for id, zone := range zones {
result[id] = zone.zone
}
@ -324,61 +323,57 @@ func (p *AWSProvider) zones(ctx context.Context) (map[string]*profiledZone, erro
log.Debug("Refreshing zones list cache")
zones := make(map[string]*profiledZone)
var profile string
var tagErr error
f := func(resp *route53.ListHostedZonesOutput, lastPage bool) (shouldContinue bool) {
for _, zone := range resp.HostedZones {
if !p.zoneIDFilter.Match(aws.StringValue(zone.Id)) {
continue
}
if !p.zoneTypeFilter.Match(zone) {
continue
}
for profile, client := range p.clients {
var tagErr error
paginator := route53.NewListHostedZonesPaginator(client, &route53.ListHostedZonesInput{})
if !p.domainFilter.Match(aws.StringValue(zone.Name)) {
if !p.zoneMatchParent {
continue
}
if !p.domainFilter.MatchParent(aws.StringValue(zone.Name)) {
continue
}
}
// Only fetch tags if a tag filter was specified
if !p.zoneTagFilter.IsEmpty() {
tags, err := p.tagsForZone(ctx, *zone.Id, profile)
if err != nil {
tagErr = err
return false
}
if !p.zoneTagFilter.Match(tags) {
continue
}
}
zones[aws.StringValue(zone.Id)] = &profiledZone{
profile: profile,
zone: zone,
}
}
return true
}
for p, client := range p.clients {
profile = p
err := client.ListHostedZonesPagesWithContext(ctx, &route53.ListHostedZonesInput{}, f)
if err != nil {
var awsErr awserr.Error
if errors.As(err, &awsErr) {
if awsErr.Code() == route53.ErrCodeThrottlingException {
log.Warnf("Skipping AWS profile %q due to provider side throttling: %v", profile, awsErr.Message())
for paginator.HasMorePages() {
resp, err := paginator.NextPage(ctx)
if err != nil {
var te *route53types.ThrottlingException
if errors.As(err, &te) {
log.Infof("Skipping AWS profile %q due to provider side throttling: %v", profile, te.ErrorMessage())
continue
}
// nothing to do here. Falling through to general error handling
return nil, provider.NewSoftError(fmt.Errorf("failed to list hosted zones: %w", err))
}
for _, zone := range resp.HostedZones {
if !p.zoneIDFilter.Match(*zone.Id) {
continue
}
if !p.zoneTypeFilter.Match(zone) {
continue
}
if !p.domainFilter.Match(*zone.Name) {
if !p.zoneMatchParent {
continue
}
if !p.domainFilter.MatchParent(*zone.Name) {
continue
}
}
// Only fetch tags if a tag filter was specified
if !p.zoneTagFilter.IsEmpty() {
tags, err := p.tagsForZone(ctx, *zone.Id, profile)
if err != nil {
tagErr = err
break
}
if !p.zoneTagFilter.Match(tags) {
continue
}
}
zones[*zone.Id] = &profiledZone{
profile: profile,
zone: &zone,
}
}
return nil, provider.NewSoftError(fmt.Errorf("failed to list hosted zones: %w", err))
}
if tagErr != nil {
return nil, provider.NewSoftError(fmt.Errorf("failed to list zones tags: %w", tagErr))
@ -386,7 +381,7 @@ func (p *AWSProvider) zones(ctx context.Context) (map[string]*profiledZone, erro
}
for _, zone := range zones {
log.Debugf("Considering zone: %s (domain: %s)", aws.StringValue(zone.zone.Id), aws.StringValue(zone.zone.Name))
log.Debugf("Considering zone: %s (domain: %s)", *zone.zone.Id, *zone.zone.Name)
}
if p.zonesCache.duration > time.Duration(0) {
@ -415,92 +410,93 @@ func (p *AWSProvider) Records(ctx context.Context) (endpoints []*endpoint.Endpoi
func (p *AWSProvider) records(ctx context.Context, zones map[string]*profiledZone) ([]*endpoint.Endpoint, error) {
endpoints := make([]*endpoint.Endpoint, 0)
f := func(resp *route53.ListResourceRecordSetsOutput, lastPage bool) (shouldContinue bool) {
for _, r := range resp.ResourceRecordSets {
newEndpoints := make([]*endpoint.Endpoint, 0)
if !p.SupportedRecordType(aws.StringValue(r.Type)) {
continue
}
var ttl endpoint.TTL
if r.TTL != nil {
ttl = endpoint.TTL(*r.TTL)
}
if len(r.ResourceRecords) > 0 {
targets := make([]string, len(r.ResourceRecords))
for idx, rr := range r.ResourceRecords {
targets[idx] = aws.StringValue(rr.Value)
}
ep := endpoint.NewEndpointWithTTL(wildcardUnescape(aws.StringValue(r.Name)), aws.StringValue(r.Type), ttl, targets...)
if aws.StringValue(r.Type) == endpoint.RecordTypeCNAME {
ep = ep.WithProviderSpecific(providerSpecificAlias, "false")
}
newEndpoints = append(newEndpoints, ep)
}
if r.AliasTarget != nil {
// Alias records don't have TTLs so provide the default to match the TXT generation
if ttl == 0 {
ttl = recordTTL
}
ep := endpoint.
NewEndpointWithTTL(wildcardUnescape(aws.StringValue(r.Name)), endpoint.RecordTypeA, ttl, aws.StringValue(r.AliasTarget.DNSName)).
WithProviderSpecific(providerSpecificEvaluateTargetHealth, fmt.Sprintf("%t", aws.BoolValue(r.AliasTarget.EvaluateTargetHealth))).
WithProviderSpecific(providerSpecificAlias, "true")
newEndpoints = append(newEndpoints, ep)
}
for _, ep := range newEndpoints {
if r.SetIdentifier != nil {
ep.SetIdentifier = aws.StringValue(r.SetIdentifier)
switch {
case r.Weight != nil:
ep.WithProviderSpecific(providerSpecificWeight, fmt.Sprintf("%d", aws.Int64Value(r.Weight)))
case r.Region != nil:
ep.WithProviderSpecific(providerSpecificRegion, aws.StringValue(r.Region))
case r.Failover != nil:
ep.WithProviderSpecific(providerSpecificFailover, aws.StringValue(r.Failover))
case r.MultiValueAnswer != nil && aws.BoolValue(r.MultiValueAnswer):
ep.WithProviderSpecific(providerSpecificMultiValueAnswer, "")
case r.GeoLocation != nil:
if r.GeoLocation.ContinentCode != nil {
ep.WithProviderSpecific(providerSpecificGeolocationContinentCode, aws.StringValue(r.GeoLocation.ContinentCode))
} else {
if r.GeoLocation.CountryCode != nil {
ep.WithProviderSpecific(providerSpecificGeolocationCountryCode, aws.StringValue(r.GeoLocation.CountryCode))
}
if r.GeoLocation.SubdivisionCode != nil {
ep.WithProviderSpecific(providerSpecificGeolocationSubdivisionCode, aws.StringValue(r.GeoLocation.SubdivisionCode))
}
}
default:
// one of the above needs to be set, otherwise SetIdentifier doesn't make sense
}
}
if r.HealthCheckId != nil {
ep.WithProviderSpecific(providerSpecificHealthCheckID, aws.StringValue(r.HealthCheckId))
}
endpoints = append(endpoints, ep)
}
}
return true
}
for _, z := range zones {
params := &route53.ListResourceRecordSetsInput{
HostedZoneId: z.zone.Id,
MaxItems: aws.String(route53PageSize),
}
client := p.clients[z.profile]
if err := client.ListResourceRecordSetsPagesWithContext(ctx, params, f); err != nil {
return nil, fmt.Errorf("failed to list resource records sets for zone %s using aws profile %q: %w", *z.zone.Id, z.profile, err)
paginator := route53.NewListResourceRecordSetsPaginator(client, &route53.ListResourceRecordSetsInput{
HostedZoneId: z.zone.Id,
MaxItems: aws.Int32(route53PageSize),
})
for paginator.HasMorePages() {
resp, err := paginator.NextPage(ctx)
if err != nil {
return nil, fmt.Errorf("failed to list resource records sets for zone %s using aws profile %q: %w", *z.zone.Id, z.profile, err)
}
for _, r := range resp.ResourceRecordSets {
newEndpoints := make([]*endpoint.Endpoint, 0)
if !p.SupportedRecordType(r.Type) {
continue
}
var ttl endpoint.TTL
if r.TTL != nil {
ttl = endpoint.TTL(*r.TTL)
}
if len(r.ResourceRecords) > 0 {
targets := make([]string, len(r.ResourceRecords))
for idx, rr := range r.ResourceRecords {
targets[idx] = *rr.Value
}
ep := endpoint.NewEndpointWithTTL(wildcardUnescape(*r.Name), string(r.Type), ttl, targets...)
if r.Type == endpoint.RecordTypeCNAME {
ep = ep.WithProviderSpecific(providerSpecificAlias, "false")
}
newEndpoints = append(newEndpoints, ep)
}
if r.AliasTarget != nil {
// Alias records don't have TTLs so provide the default to match the TXT generation
if ttl == 0 {
ttl = recordTTL
}
ep := endpoint.
NewEndpointWithTTL(wildcardUnescape(*r.Name), endpoint.RecordTypeA, ttl, *r.AliasTarget.DNSName).
WithProviderSpecific(providerSpecificEvaluateTargetHealth, fmt.Sprintf("%t", r.AliasTarget.EvaluateTargetHealth)).
WithProviderSpecific(providerSpecificAlias, "true")
newEndpoints = append(newEndpoints, ep)
}
for _, ep := range newEndpoints {
if r.SetIdentifier != nil {
ep.SetIdentifier = *r.SetIdentifier
switch {
case r.Weight != nil:
ep.WithProviderSpecific(providerSpecificWeight, fmt.Sprintf("%d", *r.Weight))
case r.Region != "":
ep.WithProviderSpecific(providerSpecificRegion, string(r.Region))
case r.Failover != "":
ep.WithProviderSpecific(providerSpecificFailover, string(r.Failover))
case r.MultiValueAnswer != nil && *r.MultiValueAnswer:
ep.WithProviderSpecific(providerSpecificMultiValueAnswer, "")
case r.GeoLocation != nil:
if r.GeoLocation.ContinentCode != nil {
ep.WithProviderSpecific(providerSpecificGeolocationContinentCode, *r.GeoLocation.ContinentCode)
} else {
if r.GeoLocation.CountryCode != nil {
ep.WithProviderSpecific(providerSpecificGeolocationCountryCode, *r.GeoLocation.CountryCode)
}
if r.GeoLocation.SubdivisionCode != nil {
ep.WithProviderSpecific(providerSpecificGeolocationSubdivisionCode, *r.GeoLocation.SubdivisionCode)
}
}
default:
// one of the above needs to be set, otherwise SetIdentifier doesn't make sense
}
}
if r.HealthCheckId != nil {
ep.WithProviderSpecific(providerSpecificHealthCheckID, *r.HealthCheckId)
}
endpoints = append(endpoints, ep)
}
}
}
}
@ -560,9 +556,9 @@ func (p *AWSProvider) createUpdateChanges(newEndpoints, oldEndpoints []*endpoint
}
combined := make(Route53Changes, 0, len(deletes)+len(creates)+len(updates))
combined = append(combined, p.newChanges(route53.ChangeActionCreate, creates)...)
combined = append(combined, p.newChanges(route53.ChangeActionUpsert, updates)...)
combined = append(combined, p.newChanges(route53.ChangeActionDelete, deletes)...)
combined = append(combined, p.newChanges(route53types.ChangeActionCreate, creates)...)
combined = append(combined, p.newChanges(route53types.ChangeActionUpsert, updates)...)
combined = append(combined, p.newChanges(route53types.ChangeActionDelete, deletes)...)
return combined
}
@ -575,7 +571,7 @@ func (p *AWSProvider) GetDomainFilter() endpoint.DomainFilterInterface {
}
zoneNames := []string(nil)
for _, z := range zones {
zoneNames = append(zoneNames, aws.StringValue(z.Name), "."+aws.StringValue(z.Name))
zoneNames = append(zoneNames, *z.Name, "."+*z.Name)
}
log.Infof("Applying provider record filter for domains: %v", zoneNames)
return endpoint.NewDomainFilter(zoneNames)
@ -591,8 +587,8 @@ func (p *AWSProvider) ApplyChanges(ctx context.Context, changes *plan.Changes) e
updateChanges := p.createUpdateChanges(changes.UpdateNew, changes.UpdateOld)
combinedChanges := make(Route53Changes, 0, len(changes.Delete)+len(changes.Create)+len(updateChanges))
combinedChanges = append(combinedChanges, p.newChanges(route53.ChangeActionCreate, changes.Create)...)
combinedChanges = append(combinedChanges, p.newChanges(route53.ChangeActionDelete, changes.Delete)...)
combinedChanges = append(combinedChanges, p.newChanges(route53types.ChangeActionCreate, changes.Create)...)
combinedChanges = append(combinedChanges, p.newChanges(route53types.ChangeActionDelete, changes.Delete)...)
combinedChanges = append(combinedChanges, updateChanges...)
return p.submitChanges(ctx, combinedChanges, zones)
@ -615,7 +611,7 @@ func (p *AWSProvider) submitChanges(ctx context.Context, changes Route53Changes,
var failedZones []string
for z, cs := range changesByZone {
log := log.WithFields(log.Fields{
"zoneName": aws.StringValue(zones[z].zone.Name),
"zoneName": *zones[z].zone.Name,
"zoneID": z,
"profile": zones[z].profile,
})
@ -634,13 +630,13 @@ func (p *AWSProvider) submitChanges(ctx context.Context, changes Route53Changes,
}
for _, c := range b {
log.Infof("Desired change: %s %s %s", *c.Action, *c.ResourceRecordSet.Name, *c.ResourceRecordSet.Type)
log.Infof("Desired change: %s %s %s", c.Action, *c.ResourceRecordSet.Name, c.ResourceRecordSet.Type)
}
if !p.dryRun {
params := &route53.ChangeResourceRecordSetsInput{
HostedZoneId: aws.String(z),
ChangeBatch: &route53.ChangeBatch{
ChangeBatch: &route53types.ChangeBatch{
Changes: b.Route53Changes(),
},
}
@ -648,8 +644,8 @@ func (p *AWSProvider) submitChanges(ctx context.Context, changes Route53Changes,
successfulChanges := 0
client := p.clients[zones[z].profile]
if _, err := client.ChangeResourceRecordSetsWithContext(ctx, params); err != nil {
log.Errorf("Failure in zone %s when submitting change batch: %v", aws.StringValue(zones[z].zone.Name), err)
if _, err := client.ChangeResourceRecordSets(ctx, params); err != nil {
log.Errorf("Failure in zone %s when submitting change batch: %v", *zones[z].zone.Name, err)
changesByOwnership := groupChangesByNameAndOwnershipRelation(b)
@ -658,12 +654,12 @@ func (p *AWSProvider) submitChanges(ctx context.Context, changes Route53Changes,
for _, changes := range changesByOwnership {
for _, c := range changes {
log.Debugf("Desired change: %s %s %s", *c.Action, *c.ResourceRecordSet.Name, *c.ResourceRecordSet.Type)
log.Debugf("Desired change: %s %s %s", c.Action, *c.ResourceRecordSet.Name, c.ResourceRecordSet.Type)
}
params.ChangeBatch = &route53.ChangeBatch{
params.ChangeBatch = &route53types.ChangeBatch{
Changes: changes.Route53Changes(),
}
if _, err := client.ChangeResourceRecordSetsWithContext(ctx, params); err != nil {
if _, err := client.ChangeResourceRecordSets(ctx, params); err != nil {
failedUpdate = true
log.Errorf("Failed submitting change (error: %v), it will be retried in a separate change batch in the next iteration", err)
p.failedChangesQueue[z] = append(p.failedChangesQueue[z], changes...)
@ -702,7 +698,7 @@ func (p *AWSProvider) submitChanges(ctx context.Context, changes Route53Changes,
}
// newChanges returns a collection of Changes based on the given records and action.
func (p *AWSProvider) newChanges(action string, endpoints []*endpoint.Endpoint) Route53Changes {
func (p *AWSProvider) newChanges(action route53types.ChangeAction, endpoints []*endpoint.Endpoint) Route53Changes {
changes := make(Route53Changes, 0, len(endpoints))
for _, endpoint := range endpoints {
@ -711,8 +707,10 @@ func (p *AWSProvider) newChanges(action string, endpoints []*endpoint.Endpoint)
if dualstack {
// make a copy of change, modify RRS type to AAAA, then add new change
rrs := *change.ResourceRecordSet
change2 := &Route53Change{Change: route53.Change{Action: change.Action, ResourceRecordSet: &rrs}}
change2.ResourceRecordSet.Type = aws.String(route53.RRTypeAaaa)
change2 := &Route53Change{
Change: route53types.Change{Action: change.Action, ResourceRecordSet: &rrs},
}
change2.ResourceRecordSet.Type = route53types.RRTypeAaaa
changes = append(changes, change2)
}
}
@ -774,11 +772,11 @@ func (p *AWSProvider) AdjustEndpoints(endpoints []*endpoint.Endpoint) ([]*endpoi
// returned Change is based on the given record by the given action, e.g.
// action=ChangeActionCreate returns a change for creation of the record and
// action=ChangeActionDelete returns a change for deletion of the record.
func (p *AWSProvider) newChange(action string, ep *endpoint.Endpoint) (*Route53Change, bool) {
func (p *AWSProvider) newChange(action route53types.ChangeAction, ep *endpoint.Endpoint) (*Route53Change, bool) {
change := &Route53Change{
Change: route53.Change{
Action: aws.String(action),
ResourceRecordSet: &route53.ResourceRecordSet{
Change: route53types.Change{
Action: action,
ResourceRecordSet: &route53types.ResourceRecordSet{
Name: aws.String(ep.DNSName),
},
},
@ -793,24 +791,24 @@ func (p *AWSProvider) newChange(action string, ep *endpoint.Endpoint) (*Route53C
if val, ok := ep.Labels[endpoint.DualstackLabelKey]; ok {
dualstack = val == "true"
}
change.ResourceRecordSet.Type = aws.String(route53.RRTypeA)
change.ResourceRecordSet.AliasTarget = &route53.AliasTarget{
change.ResourceRecordSet.Type = route53types.RRTypeA
change.ResourceRecordSet.AliasTarget = &route53types.AliasTarget{
DNSName: aws.String(ep.Targets[0]),
HostedZoneId: aws.String(cleanZoneID(targetHostedZone)),
EvaluateTargetHealth: aws.Bool(evalTargetHealth),
EvaluateTargetHealth: evalTargetHealth,
}
change.sizeBytes += len([]byte(ep.Targets[0]))
change.sizeValues += 1
} else {
change.ResourceRecordSet.Type = aws.String(ep.RecordType)
change.ResourceRecordSet.Type = route53types.RRType(ep.RecordType)
if !ep.RecordTTL.IsConfigured() {
change.ResourceRecordSet.TTL = aws.Int64(recordTTL)
} else {
change.ResourceRecordSet.TTL = aws.Int64(int64(ep.RecordTTL))
}
change.ResourceRecordSet.ResourceRecords = make([]*route53.ResourceRecord, len(ep.Targets))
change.ResourceRecordSet.ResourceRecords = make([]route53types.ResourceRecord, len(ep.Targets))
for idx, val := range ep.Targets {
change.ResourceRecordSet.ResourceRecords[idx] = &route53.ResourceRecord{
change.ResourceRecordSet.ResourceRecords[idx] = route53types.ResourceRecord{
Value: aws.String(val),
}
change.sizeBytes += len([]byte(val))
@ -818,7 +816,7 @@ func (p *AWSProvider) newChange(action string, ep *endpoint.Endpoint) (*Route53C
}
}
if action == route53.ChangeActionUpsert {
if action == route53types.ChangeActionUpsert {
// If the value of the Action element is UPSERT, each ResourceRecord element and each character in a Value
// element is counted twice
change.sizeBytes *= 2
@ -837,16 +835,16 @@ func (p *AWSProvider) newChange(action string, ep *endpoint.Endpoint) (*Route53C
change.ResourceRecordSet.Weight = aws.Int64(weight)
}
if prop, ok := ep.GetProviderSpecificProperty(providerSpecificRegion); ok {
change.ResourceRecordSet.Region = aws.String(prop)
change.ResourceRecordSet.Region = route53types.ResourceRecordSetRegion(prop)
}
if prop, ok := ep.GetProviderSpecificProperty(providerSpecificFailover); ok {
change.ResourceRecordSet.Failover = aws.String(prop)
change.ResourceRecordSet.Failover = route53types.ResourceRecordSetFailover(prop)
}
if _, ok := ep.GetProviderSpecificProperty(providerSpecificMultiValueAnswer); ok {
change.ResourceRecordSet.MultiValueAnswer = aws.Bool(true)
}
geolocation := &route53.GeoLocation{}
geolocation := &route53types.GeoLocation{}
useGeolocation := false
if prop, ok := ep.GetProviderSpecificProperty(providerSpecificGeolocationContinentCode); ok {
geolocation.ContinentCode = aws.String(prop)
@ -908,7 +906,7 @@ func groupChangesByNameAndOwnershipRelation(cs Route53Changes) map[string]Route5
for _, v := range cs {
key := v.OwnedRecord
if key == "" {
key = aws.StringValue(v.ResourceRecordSet.Name)
key = *v.ResourceRecordSet.Name
}
changesByOwnership[key] = append(changesByOwnership[key], v)
}
@ -918,8 +916,8 @@ func groupChangesByNameAndOwnershipRelation(cs Route53Changes) map[string]Route5
func (p *AWSProvider) tagsForZone(ctx context.Context, zoneID string, profile string) (map[string]string, error) {
client := p.clients[profile]
response, err := client.ListTagsForResourceWithContext(ctx, &route53.ListTagsForResourceInput{
ResourceType: aws.String("hostedzone"),
response, err := client.ListTagsForResource(ctx, &route53.ListTagsForResourceInput{
ResourceType: route53types.TagResourceTypeHostedzone,
ResourceId: aws.String(zoneID),
})
if err != nil {
@ -1006,10 +1004,10 @@ func batchChangeSet(cs Route53Changes, batchSize int, batchSizeBytes int, batchS
func sortChangesByActionNameType(cs Route53Changes) Route53Changes {
sort.SliceStable(cs, func(i, j int) bool {
if *cs[i].Action > *cs[j].Action {
if cs[i].Action > cs[j].Action {
return true
}
if *cs[i].Action < *cs[j].Action {
if cs[i].Action < cs[j].Action {
return false
}
if *cs[i].ResourceRecordSet.Name < *cs[j].ResourceRecordSet.Name {
@ -1018,7 +1016,7 @@ func sortChangesByActionNameType(cs Route53Changes) Route53Changes {
if *cs[i].ResourceRecordSet.Name > *cs[j].ResourceRecordSet.Name {
return false
}
return *cs[i].ResourceRecordSet.Type < *cs[j].ResourceRecordSet.Type
return cs[i].ResourceRecordSet.Type < cs[j].ResourceRecordSet.Type
})
return cs
@ -1029,34 +1027,34 @@ func changesByZone(zones map[string]*profiledZone, changeSet Route53Changes) map
changes := make(map[string]Route53Changes)
for _, z := range zones {
changes[aws.StringValue(z.zone.Id)] = Route53Changes{}
changes[*z.zone.Id] = Route53Changes{}
}
for _, c := range changeSet {
hostname := provider.EnsureTrailingDot(aws.StringValue(c.ResourceRecordSet.Name))
hostname := provider.EnsureTrailingDot(*c.ResourceRecordSet.Name)
zones := suitableZones(hostname, zones)
if len(zones) == 0 {
log.Debugf("Skipping record %s because no hosted zone matching record DNS Name was detected", c.String())
log.Debugf("Skipping record %s because no hosted zone matching record DNS Name was detected", *c.ResourceRecordSet.Name)
continue
}
for _, z := range zones {
if c.ResourceRecordSet.AliasTarget != nil && aws.StringValue(c.ResourceRecordSet.AliasTarget.HostedZoneId) == sameZoneAlias {
if c.ResourceRecordSet.AliasTarget != nil && *c.ResourceRecordSet.AliasTarget.HostedZoneId == sameZoneAlias {
// alias record is to be created; target needs to be in the same zone as endpoint
// if it's not, this will fail
rrset := *c.ResourceRecordSet
aliasTarget := *rrset.AliasTarget
aliasTarget.HostedZoneId = aws.String(cleanZoneID(aws.StringValue(z.zone.Id)))
aliasTarget.HostedZoneId = aws.String(cleanZoneID(*z.zone.Id))
rrset.AliasTarget = &aliasTarget
c = &Route53Change{
Change: route53.Change{
Change: route53types.Change{
Action: c.Action,
ResourceRecordSet: &rrset,
},
}
}
changes[aws.StringValue(z.zone.Id)] = append(changes[aws.StringValue(z.zone.Id)], c)
log.Debugf("Adding %s to zone %s [Id: %s]", hostname, aws.StringValue(z.zone.Name), aws.StringValue(z.zone.Id))
changes[*z.zone.Id] = append(changes[*z.zone.Id], c)
log.Debugf("Adding %s to zone %s [Id: %s]", hostname, *z.zone.Name, *z.zone.Id)
}
}
@ -1078,10 +1076,10 @@ func suitableZones(hostname string, zones map[string]*profiledZone) []*profiledZ
var publicZone *profiledZone
for _, z := range zones {
if aws.StringValue(z.zone.Name) == hostname || strings.HasSuffix(hostname, "."+aws.StringValue(z.zone.Name)) {
if z.zone.Config == nil || !aws.BoolValue(z.zone.Config.PrivateZone) {
if *z.zone.Name == hostname || strings.HasSuffix(hostname, "."+*z.zone.Name) {
if z.zone.Config == nil || !z.zone.Config.PrivateZone {
// Only select the best matching public zone
if publicZone == nil || len(aws.StringValue(z.zone.Name)) > len(aws.StringValue(publicZone.zone.Name)) {
if publicZone == nil || len(*z.zone.Name) > len(*publicZone.zone.Name) {
publicZone = z
}
} else {
@ -1156,11 +1154,11 @@ func cleanZoneID(id string) string {
return strings.TrimPrefix(id, "/hostedzone/")
}
func (p *AWSProvider) SupportedRecordType(recordType string) bool {
func (p *AWSProvider) SupportedRecordType(recordType route53types.RRType) bool {
switch recordType {
case "MX":
case route53types.RRTypeMx:
return true
default:
return provider.SupportedRecordType(recordType)
return provider.SupportedRecordType(string(recordType))
}
}

764
provider/aws/aws_test.go
View File

File diff suppressed because it is too large Load Diff

78
provider/aws/session.go → provider/aws/config.go
View File

@ -17,13 +17,16 @@ limitations under the License.
package aws
import (
"context"
"fmt"
"net/http"
"strings"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
"github.com/aws/aws-sdk-go/aws/request"
"github.com/aws/aws-sdk-go/aws/session"
awsv2 "github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/aws/retry"
"github.com/aws/aws-sdk-go-v2/config"
stscredsv2 "github.com/aws/aws-sdk-go-v2/credentials/stscreds"
"github.com/aws/aws-sdk-go-v2/service/sts"
"github.com/linki/instrumented_http"
"github.com/sirupsen/logrus"
@ -38,8 +41,8 @@ type AWSSessionConfig struct {
Profile string
}
func CreateDefaultSession(cfg *externaldns.Config) *session.Session {
result, err := newSession(
func CreateDefaultV2Config(cfg *externaldns.Config) awsv2.Config {
result, err := newV2Config(
AWSSessionConfig{
AssumeRole: cfg.AWSAssumeRole,
AssumeRoleExternalID: cfg.AWSAssumeRoleExternalID,
@ -52,24 +55,14 @@ func CreateDefaultSession(cfg *externaldns.Config) *session.Session {
return result
}
func CreateSessions(cfg *externaldns.Config) map[string]*session.Session {
result := make(map[string]*session.Session)
func CreateV2Configs(cfg *externaldns.Config) map[string]awsv2.Config {
result := make(map[string]awsv2.Config)
if len(cfg.AWSProfiles) == 0 || (len(cfg.AWSProfiles) == 1 && cfg.AWSProfiles[0] == "") {
session, err := newSession(
AWSSessionConfig{
AssumeRole: cfg.AWSAssumeRole,
AssumeRoleExternalID: cfg.AWSAssumeRoleExternalID,
APIRetries: cfg.AWSAPIRetries,
},
)
if err != nil {
logrus.Fatal(err)
}
result[defaultAWSProfile] = session
cfg := CreateDefaultV2Config(cfg)
result[defaultAWSProfile] = cfg
} else {
for _, profile := range cfg.AWSProfiles {
session, err := newSession(
cfg, err := newV2Config(
AWSSessionConfig{
AssumeRole: cfg.AWSAssumeRole,
AssumeRoleExternalID: cfg.AWSAssumeRoleExternalID,
@ -80,46 +73,47 @@ func CreateSessions(cfg *externaldns.Config) map[string]*session.Session {
if err != nil {
logrus.Fatal(err)
}
result[profile] = session
result[profile] = cfg
}
}
return result
}
func newSession(awsConfig AWSSessionConfig) (*session.Session, error) {
config := aws.NewConfig().WithMaxRetries(awsConfig.APIRetries)
config.WithHTTPClient(
instrumented_http.NewClient(config.HTTPClient, &instrumented_http.Callbacks{
func newV2Config(awsConfig AWSSessionConfig) (awsv2.Config, error) {
defaultOpts := []func(*config.LoadOptions) error{
config.WithRetryer(func() awsv2.Retryer {
return retry.AddWithMaxAttempts(retry.NewStandard(), awsConfig.APIRetries)
}),
config.WithHTTPClient(instrumented_http.NewClient(&http.Client{}, &instrumented_http.Callbacks{
PathProcessor: func(path string) string {
parts := strings.Split(path, "/")
return parts[len(parts)-1]
},
}),
)
})),
config.WithSharedConfigProfile(awsConfig.Profile),
}
session, err := session.NewSessionWithOptions(session.Options{
Config: *config,
SharedConfigState: session.SharedConfigEnable,
Profile: awsConfig.Profile,
})
cfg, err := config.LoadDefaultConfig(context.Background(), defaultOpts...)
if err != nil {
return nil, fmt.Errorf("instantiating AWS session: %w", err)
return awsv2.Config{}, fmt.Errorf("instantiating AWS config: %w", err)
}
if awsConfig.AssumeRole != "" {
stsSvc := sts.NewFromConfig(cfg)
var assumeRoleOpts []func(*stscredsv2.AssumeRoleOptions)
if awsConfig.AssumeRoleExternalID != "" {
logrus.Infof("Assuming role: %s with external id %s", awsConfig.AssumeRole, awsConfig.AssumeRoleExternalID)
session.Config.WithCredentials(stscreds.NewCredentials(session, awsConfig.AssumeRole, func(p *stscreds.AssumeRoleProvider) {
p.ExternalID = &awsConfig.AssumeRoleExternalID
}))
assumeRoleOpts = []func(*stscredsv2.AssumeRoleOptions){
func(opts *stscredsv2.AssumeRoleOptions) {
opts.ExternalID = &awsConfig.AssumeRoleExternalID
},
}
} else {
logrus.Infof("Assuming role: %s", awsConfig.AssumeRole)
session.Config.WithCredentials(stscreds.NewCredentials(session, awsConfig.AssumeRole))
}
creds := stscredsv2.NewAssumeRoleProvider(stsSvc, awsConfig.AssumeRole, assumeRoleOpts...)
cfg.Credentials = awsv2.NewCredentialsCache(creds)
}
session.Handlers.Build.PushBack(request.MakeAddToUserAgentHandler("ExternalDNS", externaldns.Version))
return session, nil
return cfg, nil
}

11
provider/aws/session_test.go → provider/aws/config_test.go
View File

@ -17,6 +17,7 @@ limitations under the License.
package aws
import (
"context"
"os"
"testing"
@ -24,7 +25,7 @@ import (
"github.com/stretchr/testify/require"
)
func Test_newSession(t *testing.T) {
func Test_newV2Config(t *testing.T) {
t.Run("should use profile from credentials file", func(t *testing.T) {
// setup
credsFile, err := prepareCredentialsFile(t)
@ -34,9 +35,9 @@ func Test_newSession(t *testing.T) {
defer os.Unsetenv("AWS_SHARED_CREDENTIALS_FILE")
// when
s, err := newSession(AWSSessionConfig{Profile: "profile2"})
cfg, err := newV2Config(AWSSessionConfig{Profile: "profile2"})
require.NoError(t, err)
creds, err := s.Config.Credentials.Get()
creds, err := cfg.Credentials.Retrieve(context.Background())
// then
assert.NoError(t, err)
@ -52,9 +53,9 @@ func Test_newSession(t *testing.T) {
defer os.Unsetenv("AWS_SECRET_ACCESS_KEY")
// when
s, err := newSession(AWSSessionConfig{})
cfg, err := newV2Config(AWSSessionConfig{})
require.NoError(t, err)
creds, err := s.Config.Credentials.Get()
creds, err := cfg.Credentials.Retrieve(context.Background())
// then
assert.NoError(t, err)

272
provider/awssd/aws_sd.go
View File

@ -24,9 +24,9 @@ import (
"regexp"
"strings"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/request"
sd "github.com/aws/aws-sdk-go/service/servicediscovery"
"github.com/aws/aws-sdk-go-v2/aws"
sd "github.com/aws/aws-sdk-go-v2/service/servicediscovery"
sdtypes "github.com/aws/aws-sdk-go-v2/service/servicediscovery/types"
log "github.com/sirupsen/logrus"
"sigs.k8s.io/external-dns/endpoint"
@ -54,16 +54,16 @@ var (
)
// AWSSDClient is the subset of the AWS Cloud Map API that we actually use. Add methods as required.
// Signatures must match exactly. Taken from https://github.com/aws/aws-sdk-go/blob/HEAD/service/servicediscovery/api.go
// Signatures must match exactly. Taken from https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/servicediscovery
type AWSSDClient interface {
CreateService(input *sd.CreateServiceInput) (*sd.CreateServiceOutput, error)
DeregisterInstance(input *sd.DeregisterInstanceInput) (*sd.DeregisterInstanceOutput, error)
DiscoverInstancesWithContext(ctx aws.Context, input *sd.DiscoverInstancesInput, opts ...request.Option) (*sd.DiscoverInstancesOutput, 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)
DeleteService(input *sd.DeleteServiceInput) (*sd.DeleteServiceOutput, error)
CreateService(ctx context.Context, params *sd.CreateServiceInput, optFns ...func(*sd.Options)) (*sd.CreateServiceOutput, error)
DeregisterInstance(ctx context.Context, params *sd.DeregisterInstanceInput, optFns ...func(*sd.Options)) (*sd.DeregisterInstanceOutput, error)
DiscoverInstances(ctx context.Context, params *sd.DiscoverInstancesInput, optFns ...func(*sd.Options)) (*sd.DiscoverInstancesOutput, error)
ListNamespaces(ctx context.Context, params *sd.ListNamespacesInput, optFns ...func(*sd.Options)) (*sd.ListNamespacesOutput, error)
ListServices(ctx context.Context, params *sd.ListServicesInput, optFns ...func(*sd.Options)) (*sd.ListServicesOutput, error)
RegisterInstance(ctx context.Context, params *sd.RegisterInstanceInput, optFns ...func(*sd.Options)) (*sd.RegisterInstanceOutput, error)
UpdateService(ctx context.Context, params *sd.UpdateServiceInput, optFns ...func(*sd.Options)) (*sd.UpdateServiceOutput, error)
DeleteService(ctx context.Context, params *sd.DeleteServiceInput, optFns ...func(*sd.Options)) (*sd.DeleteServiceOutput, error)
}
// AWSSDProvider is an implementation of Provider for AWS Cloud Map.
@ -74,7 +74,7 @@ type AWSSDProvider struct {
// only consider namespaces ending in this suffix
namespaceFilter endpoint.DomainFilter
// filter namespace by type (private or public)
namespaceTypeFilter *sd.NamespaceFilter
namespaceTypeFilter sdtypes.NamespaceFilter
// enables service without instances cleanup
cleanEmptyService bool
// filter services for removal
@ -83,7 +83,7 @@ type AWSSDProvider struct {
// NewAWSSDProvider initializes a new AWS Cloud Map based Provider.
func NewAWSSDProvider(domainFilter endpoint.DomainFilter, namespaceType string, dryRun, cleanEmptyService bool, ownerID string, client AWSSDClient) (*AWSSDProvider, error) {
provider := &AWSSDProvider{
p := &AWSSDProvider{
client: client,
dryRun: dryRun,
namespaceFilter: domainFilter,
@ -92,42 +92,42 @@ func NewAWSSDProvider(domainFilter endpoint.DomainFilter, namespaceType string,
ownerID: ownerID,
}
return provider, nil
return p, nil
}
// newSdNamespaceFilter initialized AWS SD Namespace Filter based on given string config
func newSdNamespaceFilter(namespaceTypeConfig string) *sd.NamespaceFilter {
func newSdNamespaceFilter(namespaceTypeConfig string) sdtypes.NamespaceFilter {
switch namespaceTypeConfig {
case sdNamespaceTypePublic:
return &sd.NamespaceFilter{
Name: aws.String(sd.NamespaceFilterNameType),
Values: []*string{aws.String(sd.NamespaceTypeDnsPublic)},
return sdtypes.NamespaceFilter{
Name: sdtypes.NamespaceFilterNameType,
Values: []string{string(sdtypes.NamespaceTypeDnsPublic)},
}
case sdNamespaceTypePrivate:
return &sd.NamespaceFilter{
Name: aws.String(sd.NamespaceFilterNameType),
Values: []*string{aws.String(sd.NamespaceTypeDnsPrivate)},
return sdtypes.NamespaceFilter{
Name: sdtypes.NamespaceFilterNameType,
Values: []string{string(sdtypes.NamespaceTypeDnsPrivate)},
}
default:
return nil
return sdtypes.NamespaceFilter{}
}
}
// Records returns list of all endpoints.
func (p *AWSSDProvider) Records(ctx context.Context) (endpoints []*endpoint.Endpoint, err error) {
namespaces, err := p.ListNamespaces()
namespaces, err := p.ListNamespaces(ctx)
if err != nil {
return nil, err
}
for _, ns := range namespaces {
services, err := p.ListServicesByNamespaceID(ns.Id)
services, err := p.ListServicesByNamespaceID(ctx, ns.Id)
if err != nil {
return nil, err
}
for _, srv := range services {
resp, err := p.client.DiscoverInstancesWithContext(ctx, &sd.DiscoverInstancesInput{
resp, err := p.client.DiscoverInstances(ctx, &sd.DiscoverInstancesInput{
NamespaceName: ns.Name,
ServiceName: srv.Name,
})
@ -136,8 +136,8 @@ func (p *AWSSDProvider) Records(ctx context.Context) (endpoints []*endpoint.Endp
}
if len(resp.Instances) == 0 {
if err := p.DeleteService(srv); err != nil {
log.Errorf("Failed to delete service %q, error: %s", aws.StringValue(srv.Name), err)
if err := p.DeleteService(ctx, srv); err != nil {
log.Errorf("Failed to delete service %q, error: %s", *srv.Name, err)
}
continue
}
@ -149,35 +149,35 @@ func (p *AWSSDProvider) Records(ctx context.Context) (endpoints []*endpoint.Endp
return endpoints, nil
}
func (p *AWSSDProvider) instancesToEndpoint(ns *sd.NamespaceSummary, srv *sd.Service, instances []*sd.HttpInstanceSummary) *endpoint.Endpoint {
func (p *AWSSDProvider) instancesToEndpoint(ns *sdtypes.NamespaceSummary, srv *sdtypes.Service, instances []sdtypes.HttpInstanceSummary) *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)
labels[endpoint.AWSSDDescriptionLabel] = *srv.Description
newEndpoint := &endpoint.Endpoint{
DNSName: recordName,
RecordTTL: endpoint.TTL(aws.Int64Value(srv.DnsConfig.DnsRecords[0].TTL)),
RecordTTL: endpoint.TTL(*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 {
if inst.Attributes[sdInstanceAttrCname] != "" && srv.DnsConfig.DnsRecords[0].Type == sdtypes.RecordTypeCname {
newEndpoint.RecordType = endpoint.RecordTypeCNAME
newEndpoint.Targets = append(newEndpoint.Targets, aws.StringValue(inst.Attributes[sdInstanceAttrCname]))
newEndpoint.Targets = append(newEndpoint.Targets, inst.Attributes[sdInstanceAttrCname])
// ALIAS
} else if inst.Attributes[sdInstanceAttrAlias] != nil {
} else if inst.Attributes[sdInstanceAttrAlias] != "" {
newEndpoint.RecordType = endpoint.RecordTypeCNAME
newEndpoint.Targets = append(newEndpoint.Targets, aws.StringValue(inst.Attributes[sdInstanceAttrAlias]))
newEndpoint.Targets = append(newEndpoint.Targets, inst.Attributes[sdInstanceAttrAlias])
// IP-based target
} else if inst.Attributes[sdInstanceAttrIPV4] != nil {
} else if inst.Attributes[sdInstanceAttrIPV4] != "" {
newEndpoint.RecordType = endpoint.RecordTypeA
newEndpoint.Targets = append(newEndpoint.Targets, aws.StringValue(inst.Attributes[sdInstanceAttrIPV4]))
newEndpoint.Targets = append(newEndpoint.Targets, inst.Attributes[sdInstanceAttrIPV4])
} else {
log.Warnf("Invalid instance \"%v\" found in service \"%v\"", inst, srv.Name)
}
@ -199,7 +199,7 @@ func (p *AWSSDProvider) ApplyChanges(ctx context.Context, changes *plan.Changes)
changes.Delete = append(changes.Delete, deletes...)
changes.Create = append(changes.Create, creates...)
namespaces, err := p.ListNamespaces()
namespaces, err := p.ListNamespaces(ctx)
if err != nil {
return err
}
@ -211,12 +211,12 @@ func (p *AWSSDProvider) ApplyChanges(ctx context.Context, changes *plan.Changes)
// 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)
err = p.submitDeletes(ctx, namespaces, changes.Delete)
if err != nil {
return err
}
err = p.submitCreates(namespaces, changes.Create)
err = p.submitCreates(ctx, namespaces, changes.Create)
if err != nil {
return err
}
@ -245,11 +245,11 @@ func (p *AWSSDProvider) updatesToCreates(changes *plan.Changes) (creates []*endp
return creates, deletes
}
func (p *AWSSDProvider) submitCreates(namespaces []*sd.NamespaceSummary, changes []*endpoint.Endpoint) error {
func (p *AWSSDProvider) submitCreates(ctx context.Context, namespaces []*sdtypes.NamespaceSummary, changes []*endpoint.Endpoint) error {
changesByNamespaceID := p.changesByNamespaceID(namespaces, changes)
for nsID, changeList := range changesByNamespaceID {
services, err := p.ListServicesByNamespaceID(aws.String(nsID))
services, err := p.ListServicesByNamespaceID(ctx, aws.String(nsID))
if err != nil {
return err
}
@ -260,7 +260,7 @@ func (p *AWSSDProvider) submitCreates(namespaces []*sd.NamespaceSummary, changes
srv := services[srvName]
if srv == nil {
// when service is missing create a new one
srv, err = p.CreateService(&nsID, &srvName, ch)
srv, err = p.CreateService(ctx, &nsID, &srvName, ch)
if err != nil {
return err
}
@ -268,13 +268,13 @@ func (p *AWSSDProvider) submitCreates(namespaces []*sd.NamespaceSummary, changes
services[*srv.Name] = srv
} else if ch.RecordTTL.IsConfigured() && *srv.DnsConfig.DnsRecords[0].TTL != int64(ch.RecordTTL) {
// update service when TTL differ
err = p.UpdateService(srv, ch)
err = p.UpdateService(ctx, srv, ch)
if err != nil {
return err
}
}
err = p.RegisterInstance(srv, ch)
err = p.RegisterInstance(ctx, srv, ch)
if err != nil {
return err
}
@ -284,11 +284,11 @@ func (p *AWSSDProvider) submitCreates(namespaces []*sd.NamespaceSummary, changes
return nil
}
func (p *AWSSDProvider) submitDeletes(namespaces []*sd.NamespaceSummary, changes []*endpoint.Endpoint) error {
func (p *AWSSDProvider) submitDeletes(ctx context.Context, namespaces []*sdtypes.NamespaceSummary, changes []*endpoint.Endpoint) error {
changesByNamespaceID := p.changesByNamespaceID(namespaces, changes)
for nsID, changeList := range changesByNamespaceID {
services, err := p.ListServicesByNamespaceID(aws.String(nsID))
services, err := p.ListServicesByNamespaceID(ctx, aws.String(nsID))
if err != nil {
return err
}
@ -302,7 +302,7 @@ func (p *AWSSDProvider) submitDeletes(namespaces []*sd.NamespaceSummary, changes
return fmt.Errorf("service \"%s\" is missing when trying to delete \"%v\"", srvName, hostname)
}
err := p.DeregisterInstance(srv, ch)
err := p.DeregisterInstance(ctx, srv, ch)
if err != nil {
return err
}
@ -313,53 +313,51 @@ func (p *AWSSDProvider) submitDeletes(namespaces []*sd.NamespaceSummary, changes
}
// ListNamespaces returns all namespaces matching defined namespace filter
func (p *AWSSDProvider) ListNamespaces() ([]*sd.NamespaceSummary, error) {
namespaces := make([]*sd.NamespaceSummary, 0)
func (p *AWSSDProvider) ListNamespaces(ctx context.Context) ([]*sdtypes.NamespaceSummary, error) {
namespaces := make([]*sdtypes.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)
paginator := sd.NewListNamespacesPaginator(p.client, &sd.ListNamespacesInput{
Filters: []sdtypes.NamespaceFilter{p.namespaceTypeFilter},
})
for paginator.HasMorePages() {
resp, err := paginator.NextPage(ctx)
if err != nil {
return nil, err
}
return true
}
err := p.client.ListNamespacesPages(&sd.ListNamespacesInput{
Filters: []*sd.NamespaceFilter{p.namespaceTypeFilter},
}, f)
if err != nil {
return nil, err
for _, ns := range resp.Namespaces {
if !p.namespaceFilter.Match(*ns.Name) {
continue
}
namespaces = append(namespaces, &ns)
}
}
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) {
services := make([]*sd.ServiceSummary, 0)
// ListServicesByNamespaceID returns list of services in given namespace.
func (p *AWSSDProvider) ListServicesByNamespaceID(ctx context.Context, namespaceID *string) (map[string]*sdtypes.Service, error) {
services := make([]sdtypes.ServiceSummary, 0)
f := func(resp *sd.ListServicesOutput, lastPage bool) bool {
services = append(services, resp.Services...)
return true
}
err := p.client.ListServicesPages(&sd.ListServicesInput{
Filters: []*sd.ServiceFilter{{
Name: aws.String(sd.ServiceFilterNameNamespaceId),
Values: []*string{namespaceID},
paginator := sd.NewListServicesPaginator(p.client, &sd.ListServicesInput{
Filters: []sdtypes.ServiceFilter{{
Name: sdtypes.ServiceFilterNameNamespaceId,
Values: []string{*namespaceID},
}},
MaxResults: aws.Int64(100),
}, f)
if err != nil {
return nil, err
MaxResults: aws.Int32(100),
})
for paginator.HasMorePages() {
resp, err := paginator.NextPage(ctx)
if err != nil {
return nil, err
}
services = append(services, resp.Services...)
}
servicesMap := make(map[string]*sd.Service)
servicesMap := make(map[string]*sdtypes.Service)
for _, serviceSummary := range services {
service := &sd.Service{
service := &sdtypes.Service{
Arn: serviceSummary.Arn,
CreateDate: serviceSummary.CreateDate,
Description: serviceSummary.Description,
@ -373,13 +371,13 @@ func (p *AWSSDProvider) ListServicesByNamespaceID(namespaceID *string) (map[stri
Type: serviceSummary.Type,
}
servicesMap[aws.StringValue(service.Name)] = service
servicesMap[*service.Name] = service
}
return servicesMap, 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) {
func (p *AWSSDProvider) CreateService(ctx context.Context, namespaceID *string, srvName *string, ep *endpoint.Endpoint) (*sdtypes.Service, error) {
log.Infof("Creating a new service \"%s\" in \"%s\" namespace", *srvName, *namespaceID)
srvType := p.serviceTypeFromEndpoint(ep)
@ -391,13 +389,13 @@ func (p *AWSSDProvider) CreateService(namespaceID *string, srvName *string, ep *
}
if !p.dryRun {
out, err := p.client.CreateService(&sd.CreateServiceInput{
out, err := p.client.CreateService(ctx, &sd.CreateServiceInput{
Name: srvName,
Description: aws.String(ep.Labels[endpoint.AWSSDDescriptionLabel]),
DnsConfig: &sd.DnsConfig{
RoutingPolicy: aws.String(routingPolicy),
DnsRecords: []*sd.DnsRecord{{
Type: aws.String(srvType),
DnsConfig: &sdtypes.DnsConfig{
RoutingPolicy: routingPolicy,
DnsRecords: []sdtypes.DnsRecord{{
Type: srvType,
TTL: aws.Int64(ttl),
}},
},
@ -411,11 +409,11 @@ func (p *AWSSDProvider) CreateService(namespaceID *string, srvName *string, ep *
}
// 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
return &sdtypes.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 {
func (p *AWSSDProvider) UpdateService(ctx context.Context, service *sdtypes.Service, ep *endpoint.Endpoint) error {
log.Infof("Updating service \"%s\"", *service.Name)
srvType := p.serviceTypeFromEndpoint(ep)
@ -426,13 +424,13 @@ func (p *AWSSDProvider) UpdateService(service *sd.Service, ep *endpoint.Endpoint
}
if !p.dryRun {
_, err := p.client.UpdateService(&sd.UpdateServiceInput{
_, err := p.client.UpdateService(ctx, &sd.UpdateServiceInput{
Id: service.Id,
Service: &sd.ServiceChange{
Service: &sdtypes.ServiceChange{
Description: aws.String(ep.Labels[endpoint.AWSSDDescriptionLabel]),
DnsConfig: &sd.DnsConfigChange{
DnsRecords: []*sd.DnsRecord{{
Type: aws.String(srvType),
DnsConfig: &sdtypes.DnsConfigChange{
DnsRecords: []sdtypes.DnsRecord{{
Type: srvType,
TTL: aws.Int64(ttl),
}},
},
@ -447,7 +445,7 @@ func (p *AWSSDProvider) UpdateService(service *sd.Service, ep *endpoint.Endpoint
}
// DeleteService deletes empty Service from AWS API if its owner id match
func (p *AWSSDProvider) DeleteService(service *sd.Service) error {
func (p *AWSSDProvider) DeleteService(ctx context.Context, service *sdtypes.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
@ -455,39 +453,39 @@ func (p *AWSSDProvider) DeleteService(service *sd.Service) error {
label[endpoint.OwnerLabelKey] = p.ownerID
label[endpoint.AWSSDDescriptionLabel] = label.SerializePlain(false)
if strings.HasPrefix(aws.StringValue(service.Description), label[endpoint.AWSSDDescriptionLabel]) {
if strings.HasPrefix(*service.Description, label[endpoint.AWSSDDescriptionLabel]) {
log.Infof("Deleting service \"%s\"", *service.Name)
_, err := p.client.DeleteService(&sd.DeleteServiceInput{
_, err := p.client.DeleteService(ctx, &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])
log.Debugf("Skipping service removal %s because owner id does not match, found: \"%s\", required: \"%s\"", *service.Name, *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 {
func (p *AWSSDProvider) RegisterInstance(ctx context.Context, service *sdtypes.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)
attr := make(map[string]string)
if ep.RecordType == endpoint.RecordTypeCNAME {
if p.isAWSLoadBalancer(target) {
attr[sdInstanceAttrAlias] = aws.String(target)
attr[sdInstanceAttrAlias] = target
} else {
attr[sdInstanceAttrCname] = aws.String(target)
attr[sdInstanceAttrCname] = target
}
} else if ep.RecordType == endpoint.RecordTypeA {
attr[sdInstanceAttrIPV4] = aws.String(target)
attr[sdInstanceAttrIPV4] = target
} else {
return fmt.Errorf("invalid endpoint type (%v)", ep)
}
if !p.dryRun {
_, err := p.client.RegisterInstance(&sd.RegisterInstanceInput{
_, err := p.client.RegisterInstance(ctx, &sd.RegisterInstanceInput{
ServiceId: service.Id,
Attributes: attr,
InstanceId: aws.String(p.targetToInstanceID(target)),
@ -502,12 +500,12 @@ func (p *AWSSDProvider) RegisterInstance(service *sd.Service, ep *endpoint.Endpo
}
// DeregisterInstance removes an instance from given service.
func (p *AWSSDProvider) DeregisterInstance(service *sd.Service, ep *endpoint.Endpoint) error {
func (p *AWSSDProvider) DeregisterInstance(ctx context.Context, service *sdtypes.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{
_, err := p.client.DeregisterInstance(ctx, &sd.DeregisterInstanceInput{
InstanceId: aws.String(p.targetToInstanceID(target)),
ServiceId: service.Id,
})
@ -531,43 +529,7 @@ func (p *AWSSDProvider) targetToInstanceID(target string) string {
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{
Arn: service.Arn,
CreateDate: service.CreateDate,
Description: service.Description,
DnsConfig: service.DnsConfig,
HealthCheckConfig: service.HealthCheckConfig,
HealthCheckCustomConfig: service.HealthCheckCustomConfig,
Id: service.Id,
InstanceCount: service.InstanceCount,
Name: service.Name,
Type: service.Type,
}
}
func (p *AWSSDProvider) changesByNamespaceID(namespaces []*sd.NamespaceSummary, changes []*endpoint.Endpoint) map[string][]*endpoint.Endpoint {
func (p *AWSSDProvider) changesByNamespaceID(namespaces []*sdtypes.NamespaceSummary, changes []*endpoint.Endpoint) map[string][]*endpoint.Endpoint {
changesByNsID := make(map[string][]*endpoint.Endpoint)
for _, ns := range namespaces {
@ -600,8 +562,8 @@ func (p *AWSSDProvider) changesByNamespaceID(namespaces []*sd.NamespaceSummary,
}
// returns list of all namespaces matching given hostname
func matchingNamespaces(hostname string, namespaces []*sd.NamespaceSummary) []*sd.NamespaceSummary {
matchingNamespaces := make([]*sd.NamespaceSummary, 0)
func matchingNamespaces(hostname string, namespaces []*sdtypes.NamespaceSummary) []*sdtypes.NamespaceSummary {
matchingNamespaces := make([]*sdtypes.NamespaceSummary, 0)
for _, ns := range namespaces {
if *ns.Name == hostname {
@ -621,26 +583,26 @@ func (p *AWSSDProvider) parseHostname(hostname string) (namespace string, servic
}
// determine service routing policy based on endpoint type
func (p *AWSSDProvider) routingPolicyFromEndpoint(ep *endpoint.Endpoint) string {
func (p *AWSSDProvider) routingPolicyFromEndpoint(ep *endpoint.Endpoint) sdtypes.RoutingPolicy {
if ep.RecordType == endpoint.RecordTypeA {
return sd.RoutingPolicyMultivalue
return sdtypes.RoutingPolicyMultivalue
}
return sd.RoutingPolicyWeighted
return sdtypes.RoutingPolicyWeighted
}
// determine service type (A, CNAME) from given endpoint
func (p *AWSSDProvider) serviceTypeFromEndpoint(ep *endpoint.Endpoint) string {
func (p *AWSSDProvider) serviceTypeFromEndpoint(ep *endpoint.Endpoint) sdtypes.RecordType {
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
// ALIAS target uses DNS record of type A
return sdtypes.RecordTypeA
}
return sd.RecordTypeCname
return sdtypes.RecordTypeCname
}
return sd.RecordTypeA
return sdtypes.RecordTypeA
}
// determine if a given hostname belongs to an AWS load balancer

437
provider/awssd/aws_sd_test.go
View File

@ -25,9 +25,9 @@ import (
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/request"
sd "github.com/aws/aws-sdk-go/service/servicediscovery"
"github.com/aws/aws-sdk-go-v2/aws"
sd "github.com/aws/aws-sdk-go-v2/service/servicediscovery"
sdtypes "github.com/aws/aws-sdk-go-v2/service/servicediscovery/types"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@ -45,17 +45,17 @@ var (
type AWSSDClientStub struct {
// map[namespace_id]namespace
namespaces map[string]*sd.Namespace
namespaces map[string]*sdtypes.Namespace
// map[namespace_id] => map[service_id]instance
services map[string]map[string]*sd.Service
services map[string]map[string]*sdtypes.Service
// map[service_id] => map[inst_id]instance
instances map[string]map[string]*sd.Instance
instances map[string]map[string]*sdtypes.Instance
}
func (s *AWSSDClientStub) CreateService(input *sd.CreateServiceInput) (*sd.CreateServiceOutput, error) {
srv := &sd.Service{
func (s *AWSSDClientStub) CreateService(ctx context.Context, input *sd.CreateServiceInput, optFns ...func(*sd.Options)) (*sd.CreateServiceOutput, error) {
srv := &sdtypes.Service{
Id: aws.String(strconv.Itoa(rand.Intn(10000))),
DnsConfig: input.DnsConfig,
Name: input.Name,
@ -66,7 +66,7 @@ func (s *AWSSDClientStub) CreateService(input *sd.CreateServiceInput) (*sd.Creat
nsServices, ok := s.services[*input.NamespaceId]
if !ok {
nsServices = make(map[string]*sd.Service)
nsServices = make(map[string]*sdtypes.Service)
s.services[*input.NamespaceId] = nsServices
}
nsServices[*srv.Id] = srv
@ -76,14 +76,14 @@ func (s *AWSSDClientStub) CreateService(input *sd.CreateServiceInput) (*sd.Creat
}, nil
}
func (s *AWSSDClientStub) DeregisterInstance(input *sd.DeregisterInstanceInput) (*sd.DeregisterInstanceOutput, error) {
func (s *AWSSDClientStub) DeregisterInstance(ctx context.Context, input *sd.DeregisterInstanceInput, optFns ...func(options *sd.Options)) (*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) {
func (s *AWSSDClientStub) GetService(ctx context.Context, input *sd.GetServiceInput, optFns ...func(options *sd.Options)) (*sd.GetServiceOutput, error) {
for _, entry := range s.services {
srv, ok := entry[*input.Id]
if ok {
@ -96,18 +96,18 @@ func (s *AWSSDClientStub) GetService(input *sd.GetServiceInput) (*sd.GetServiceO
return nil, errors.New("service not found")
}
func (s *AWSSDClientStub) DiscoverInstancesWithContext(ctx context.Context, input *sd.DiscoverInstancesInput, opts ...request.Option) (*sd.DiscoverInstancesOutput, error) {
instances := make([]*sd.HttpInstanceSummary, 0)
func (s *AWSSDClientStub) DiscoverInstances(ctx context.Context, input *sd.DiscoverInstancesInput, opts ...func(options *sd.Options)) (*sd.DiscoverInstancesOutput, error) {
instances := make([]sdtypes.HttpInstanceSummary, 0)
var foundNs bool
for _, ns := range s.namespaces {
if aws.StringValue(ns.Name) == aws.StringValue(input.NamespaceName) {
if *ns.Name == *input.NamespaceName {
foundNs = true
for _, srv := range s.services[*ns.Id] {
if aws.StringValue(srv.Name) == aws.StringValue(input.ServiceName) {
if *srv.Name == *input.ServiceName {
for _, inst := range s.instances[*srv.Id] {
instances = append(instances, instanceToHTTPInstanceSummary(inst))
instances = append(instances, *instanceToHTTPInstanceSummary(inst))
}
}
}
@ -123,57 +123,50 @@ func (s *AWSSDClientStub) DiscoverInstancesWithContext(ctx context.Context, inpu
}, nil
}
func (s *AWSSDClientStub) ListNamespacesPages(input *sd.ListNamespacesInput, fn func(*sd.ListNamespacesOutput, bool) bool) error {
namespaces := make([]*sd.NamespaceSummary, 0)
filter := input.Filters[0]
func (s *AWSSDClientStub) ListNamespaces(ctx context.Context, input *sd.ListNamespacesInput, optFns ...func(options *sd.Options)) (*sd.ListNamespacesOutput, error) {
namespaces := make([]sdtypes.NamespaceSummary, 0)
for _, ns := range s.namespaces {
if filter != nil && *filter.Name == sd.NamespaceFilterNameType {
if *ns.Type != *filter.Values[0] {
if len(input.Filters) > 0 && input.Filters[0].Name == sdtypes.NamespaceFilterNameType {
if ns.Type != sdtypes.NamespaceType(input.Filters[0].Values[0]) {
// skip namespaces not matching filter
continue
}
}
namespaces = append(namespaces, namespaceToNamespaceSummary(ns))
namespaces = append(namespaces, *namespaceToNamespaceSummary(ns))
}
fn(&sd.ListNamespacesOutput{
return &sd.ListNamespacesOutput{
Namespaces: namespaces,
}, true)
return nil
}, nil
}
func (s *AWSSDClientStub) ListServicesPages(input *sd.ListServicesInput, fn func(*sd.ListServicesOutput, bool) bool) error {
services := make([]*sd.ServiceSummary, 0)
func (s *AWSSDClientStub) ListServices(ctx context.Context, input *sd.ListServicesInput, optFns ...func(options *sd.Options)) (*sd.ListServicesOutput, error) {
services := make([]sdtypes.ServiceSummary, 0)
// get namespace filter
filter := input.Filters[0]
if filter == nil || *filter.Name != sd.ServiceFilterNameNamespaceId {
return errors.New("missing namespace filter")
if len(input.Filters) == 0 || input.Filters[0].Name != sdtypes.ServiceFilterNameNamespaceId {
return nil, errors.New("missing namespace filter")
}
nsID := filter.Values[0]
nsID := input.Filters[0].Values[0]
for _, srv := range s.services[*nsID] {
services = append(services, serviceToServiceSummary(srv))
for _, srv := range s.services[nsID] {
services = append(services, *serviceToServiceSummary(srv))
}
fn(&sd.ListServicesOutput{
return &sd.ListServicesOutput{
Services: services,
}, true)
return nil
}, nil
}
func (s *AWSSDClientStub) RegisterInstance(input *sd.RegisterInstanceInput) (*sd.RegisterInstanceOutput, error) {
func (s *AWSSDClientStub) RegisterInstance(ctx context.Context, input *sd.RegisterInstanceInput, optFns ...func(options *sd.Options)) (*sd.RegisterInstanceOutput, error) {
srvInstances, ok := s.instances[*input.ServiceId]
if !ok {
srvInstances = make(map[string]*sd.Instance)
srvInstances = make(map[string]*sdtypes.Instance)
s.instances[*input.ServiceId] = srvInstances
}
srvInstances[*input.InstanceId] = &sd.Instance{
srvInstances[*input.InstanceId] = &sdtypes.Instance{
Id: input.InstanceId,
Attributes: input.Attributes,
CreatorRequestId: input.CreatorRequestId,
@ -182,8 +175,8 @@ func (s *AWSSDClientStub) RegisterInstance(input *sd.RegisterInstanceInput) (*sd
return &sd.RegisterInstanceOutput{}, nil
}
func (s *AWSSDClientStub) UpdateService(input *sd.UpdateServiceInput) (*sd.UpdateServiceOutput, error) {
out, err := s.GetService(&sd.GetServiceInput{Id: input.Id})
func (s *AWSSDClientStub) UpdateService(ctx context.Context, input *sd.UpdateServiceInput, optFns ...func(options *sd.Options)) (*sd.UpdateServiceOutput, error) {
out, err := s.GetService(ctx, &sd.GetServiceInput{Id: input.Id})
if err != nil {
return nil, err
}
@ -197,8 +190,8 @@ func (s *AWSSDClientStub) UpdateService(input *sd.UpdateServiceInput) (*sd.Updat
return &sd.UpdateServiceOutput{}, nil
}
func (s *AWSSDClientStub) DeleteService(input *sd.DeleteServiceInput) (*sd.DeleteServiceOutput, error) {
out, err := s.GetService(&sd.GetServiceInput{Id: input.Id})
func (s *AWSSDClientStub) DeleteService(ctx context.Context, input *sd.DeleteServiceInput, optFns ...func(options *sd.Options)) (*sd.DeleteServiceOutput, error) {
out, err := s.GetService(ctx, &sd.GetServiceInput{Id: input.Id})
if err != nil {
return nil, err
}
@ -221,39 +214,69 @@ func newTestAWSSDProvider(api AWSSDClient, domainFilter endpoint.DomainFilter, n
}
}
// nolint: deadcode
// used for unit test
func instanceToHTTPInstanceSummary(instance *sd.Instance) *sd.HttpInstanceSummary {
func instanceToHTTPInstanceSummary(instance *sdtypes.Instance) *sdtypes.HttpInstanceSummary {
if instance == nil {
return nil
}
return &sd.HttpInstanceSummary{
return &sdtypes.HttpInstanceSummary{
InstanceId: instance.Id,
Attributes: instance.Attributes,
}
}
func namespaceToNamespaceSummary(namespace *sdtypes.Namespace) *sdtypes.NamespaceSummary {
if namespace == nil {
return nil
}
return &sdtypes.NamespaceSummary{
Id: namespace.Id,
Type: namespace.Type,
Name: namespace.Name,
Arn: namespace.Arn,
}
}
func serviceToServiceSummary(service *sdtypes.Service) *sdtypes.ServiceSummary {
if service == nil {
return nil
}
return &sdtypes.ServiceSummary{
Arn: service.Arn,
CreateDate: service.CreateDate,
Description: service.Description,
DnsConfig: service.DnsConfig,
HealthCheckConfig: service.HealthCheckConfig,
HealthCheckCustomConfig: service.HealthCheckCustomConfig,
Id: service.Id,
InstanceCount: service.InstanceCount,
Name: service.Name,
Type: service.Type,
}
}
func TestAWSSDProvider_Records(t *testing.T) {
namespaces := map[string]*sd.Namespace{
namespaces := map[string]*sdtypes.Namespace{
"private": {
Id: aws.String("private"),
Name: aws.String("private.com"),
Type: aws.String(sd.NamespaceTypeDnsPrivate),
Type: sdtypes.NamespaceTypeDnsPrivate,
},
}
services := map[string]map[string]*sd.Service{
services := map[string]map[string]*sdtypes.Service{
"private": {
"a-srv": {
Id: aws.String("a-srv"),
Name: aws.String("service1"),
NamespaceId: aws.String("private"),
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),
DnsConfig: &sdtypes.DnsConfig{
RoutingPolicy: sdtypes.RoutingPolicyWeighted,
DnsRecords: []sdtypes.DnsRecord{{
Type: sdtypes.RecordTypeA,
TTL: aws.Int64(100),
}},
},
@ -261,12 +284,12 @@ func TestAWSSDProvider_Records(t *testing.T) {
"alias-srv": {
Id: aws.String("alias-srv"),
Name: aws.String("service2"),
NamespaceId: aws.String("private"),
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),
DnsConfig: &sdtypes.DnsConfig{
RoutingPolicy: sdtypes.RoutingPolicyWeighted,
DnsRecords: []sdtypes.DnsRecord{{
Type: sdtypes.RecordTypeA,
TTL: aws.Int64(100),
}},
},
@ -274,12 +297,12 @@ func TestAWSSDProvider_Records(t *testing.T) {
"cname-srv": {
Id: aws.String("cname-srv"),
Name: aws.String("service3"),
NamespaceId: aws.String("private"),
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),
DnsConfig: &sdtypes.DnsConfig{
RoutingPolicy: sdtypes.RoutingPolicyWeighted,
DnsRecords: []sdtypes.DnsRecord{{
Type: sdtypes.RecordTypeCname,
TTL: aws.Int64(80),
}},
},
@ -287,34 +310,34 @@ func TestAWSSDProvider_Records(t *testing.T) {
},
}
instances := map[string]map[string]*sd.Instance{
instances := map[string]map[string]*sdtypes.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"),
Attributes: map[string]string{
sdInstanceAttrIPV4: "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"),
Attributes: map[string]string{
sdInstanceAttrIPV4: "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"),
Attributes: map[string]string{
sdInstanceAttrAlias: "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"),
Attributes: map[string]string{
sdInstanceAttrCname: "cname.target.com",
},
},
},
@ -340,18 +363,18 @@ func TestAWSSDProvider_Records(t *testing.T) {
}
func TestAWSSDProvider_ApplyChanges(t *testing.T) {
namespaces := map[string]*sd.Namespace{
namespaces := map[string]*sdtypes.Namespace{
"private": {
Id: aws.String("private"),
Name: aws.String("private.com"),
Type: aws.String(sd.NamespaceTypeDnsPrivate),
Type: sdtypes.NamespaceTypeDnsPrivate,
},
}
api := &AWSSDClientStub{
namespaces: namespaces,
services: make(map[string]map[string]*sd.Service),
instances: make(map[string]map[string]*sd.Instance),
services: make(map[string]map[string]*sdtypes.Service),
instances: make(map[string]map[string]*sdtypes.Instance),
}
expectedEndpoints := []*endpoint.Endpoint{
@ -371,7 +394,7 @@ func TestAWSSDProvider_ApplyChanges(t *testing.T) {
// make sure services were created
assert.Len(t, api.services["private"], 3)
existingServices, _ := provider.ListServicesByNamespaceID(namespaces["private"].Id)
existingServices, _ := provider.ListServicesByNamespaceID(context.Background(), namespaces["private"].Id)
assert.NotNil(t, existingServices["service1"])
assert.NotNil(t, existingServices["service2"])
assert.NotNil(t, existingServices["service3"])
@ -392,16 +415,16 @@ func TestAWSSDProvider_ApplyChanges(t *testing.T) {
}
func TestAWSSDProvider_ListNamespaces(t *testing.T) {
namespaces := map[string]*sd.Namespace{
namespaces := map[string]*sdtypes.Namespace{
"private": {
Id: aws.String("private"),
Name: aws.String("private.com"),
Type: aws.String(sd.NamespaceTypeDnsPrivate),
Type: sdtypes.NamespaceTypeDnsPrivate,
},
"public": {
Id: aws.String("public"),
Name: aws.String("public.com"),
Type: aws.String(sd.NamespaceTypeDnsPublic),
Type: sdtypes.NamespaceTypeDnsPublic,
},
}
@ -413,20 +436,20 @@ func TestAWSSDProvider_ListNamespaces(t *testing.T) {
msg string
domainFilter endpoint.DomainFilter
namespaceTypeFilter string
expectedNamespaces []*sd.NamespaceSummary
expectedNamespaces []*sdtypes.NamespaceSummary
}{
{"public filter", endpoint.NewDomainFilter([]string{}), "public", []*sd.NamespaceSummary{namespaceToNamespaceSummary(namespaces["public"])}},
{"private filter", endpoint.NewDomainFilter([]string{}), "private", []*sd.NamespaceSummary{namespaceToNamespaceSummary(namespaces["private"])}},
{"domain filter", endpoint.NewDomainFilter([]string{"public.com"}), "", []*sd.NamespaceSummary{namespaceToNamespaceSummary(namespaces["public"])}},
{"non-existing domain", endpoint.NewDomainFilter([]string{"xxx.com"}), "", []*sd.NamespaceSummary{}},
{"public filter", endpoint.NewDomainFilter([]string{}), "public", []*sdtypes.NamespaceSummary{namespaceToNamespaceSummary(namespaces["public"])}},
{"private filter", endpoint.NewDomainFilter([]string{}), "private", []*sdtypes.NamespaceSummary{namespaceToNamespaceSummary(namespaces["private"])}},
{"domain filter", endpoint.NewDomainFilter([]string{"public.com"}), "", []*sdtypes.NamespaceSummary{namespaceToNamespaceSummary(namespaces["public"])}},
{"non-existing domain", endpoint.NewDomainFilter([]string{"xxx.com"}), "", []*sdtypes.NamespaceSummary{}},
} {
provider := newTestAWSSDProvider(api, tc.domainFilter, tc.namespaceTypeFilter, "")
result, err := provider.ListNamespaces()
result, err := provider.ListNamespaces(context.Background())
require.NoError(t, err)
expectedMap := make(map[string]*sd.NamespaceSummary)
resultMap := make(map[string]*sd.NamespaceSummary)
expectedMap := make(map[string]*sdtypes.NamespaceSummary)
resultMap := make(map[string]*sdtypes.NamespaceSummary)
for _, ns := range tc.expectedNamespaces {
expectedMap[*ns.Id] = ns
}
@ -441,20 +464,20 @@ func TestAWSSDProvider_ListNamespaces(t *testing.T) {
}
func TestAWSSDProvider_ListServicesByNamespace(t *testing.T) {
namespaces := map[string]*sd.Namespace{
namespaces := map[string]*sdtypes.Namespace{
"private": {
Id: aws.String("private"),
Name: aws.String("private.com"),
Type: aws.String(sd.NamespaceTypeDnsPrivate),
Type: sdtypes.NamespaceTypeDnsPrivate,
},
"public": {
Id: aws.String("public"),
Name: aws.String("public.com"),
Type: aws.String(sd.NamespaceTypeDnsPublic),
Type: sdtypes.NamespaceTypeDnsPublic,
},
}
services := map[string]map[string]*sd.Service{
services := map[string]map[string]*sdtypes.Service{
"private": {
"srv1": {
Id: aws.String("srv1"),
@ -482,48 +505,52 @@ func TestAWSSDProvider_ListServicesByNamespace(t *testing.T) {
}
for _, tc := range []struct {
expectedServices map[string]*sd.Service
expectedServices map[string]*sdtypes.Service
}{
{map[string]*sd.Service{"service1": services["private"]["srv1"], "service2": services["private"]["srv2"]}},
{map[string]*sdtypes.Service{"service1": services["private"]["srv1"], "service2": services["private"]["srv2"]}},
} {
provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
result, err := provider.ListServicesByNamespaceID(namespaces["private"].Id)
result, err := provider.ListServicesByNamespaceID(context.Background(), namespaces["private"].Id)
require.NoError(t, err)
assert.Equal(t, tc.expectedServices, result)
}
}
func TestAWSSDProvider_CreateService(t *testing.T) {
namespaces := map[string]*sd.Namespace{
namespaces := map[string]*sdtypes.Namespace{
"private": {
Id: aws.String("private"),
Name: aws.String("private.com"),
Type: aws.String(sd.NamespaceTypeDnsPrivate),
Type: sdtypes.NamespaceTypeDnsPrivate,
},
}
api := &AWSSDClientStub{
namespaces: namespaces,
services: make(map[string]map[string]*sd.Service),
services: make(map[string]map[string]*sdtypes.Service),
}
expectedServices := make(map[string]*sd.Service)
expectedServices := make(map[string]*sdtypes.Service)
provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
// A type
provider.CreateService(aws.String("private"), aws.String("A-srv"), &endpoint.Endpoint{
provider.CreateService(context.Background(), aws.String("private"), aws.String("A-srv"), &endpoint.Endpoint{
Labels: map[string]string{
endpoint.AWSSDDescriptionLabel: "A-srv",
},
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{
RoutingPolicy: aws.String(sd.RoutingPolicyMultivalue),
DnsRecords: []*sd.DnsRecord{{
Type: aws.String(sd.RecordTypeA),
expectedServices["A-srv"] = &sdtypes.Service{
Name: aws.String("A-srv"),
Description: aws.String("A-srv"),
DnsConfig: &sdtypes.DnsConfig{
RoutingPolicy: sdtypes.RoutingPolicyMultivalue,
DnsRecords: []sdtypes.DnsRecord{{
Type: sdtypes.RecordTypeA,
TTL: aws.Int64(60),
}},
},
@ -531,17 +558,21 @@ func TestAWSSDProvider_CreateService(t *testing.T) {
}
// CNAME type
provider.CreateService(aws.String("private"), aws.String("CNAME-srv"), &endpoint.Endpoint{
provider.CreateService(context.Background(), aws.String("private"), aws.String("CNAME-srv"), &endpoint.Endpoint{
Labels: map[string]string{
endpoint.AWSSDDescriptionLabel: "CNAME-srv",
},
RecordType: endpoint.RecordTypeCNAME,
RecordTTL: 80,
Targets: endpoint.Targets{"cname.target.com"},
})
expectedServices["CNAME-srv"] = &sd.Service{
Name: aws.String("CNAME-srv"),
DnsConfig: &sd.DnsConfig{
RoutingPolicy: aws.String(sd.RoutingPolicyWeighted),
DnsRecords: []*sd.DnsRecord{{
Type: aws.String(sd.RecordTypeCname),
expectedServices["CNAME-srv"] = &sdtypes.Service{
Name: aws.String("CNAME-srv"),
Description: aws.String("CNAME-srv"),
DnsConfig: &sdtypes.DnsConfig{
RoutingPolicy: sdtypes.RoutingPolicyWeighted,
DnsRecords: []sdtypes.DnsRecord{{
Type: sdtypes.RecordTypeCname,
TTL: aws.Int64(80),
}},
},
@ -549,17 +580,21 @@ func TestAWSSDProvider_CreateService(t *testing.T) {
}
// ALIAS type
provider.CreateService(aws.String("private"), aws.String("ALIAS-srv"), &endpoint.Endpoint{
provider.CreateService(context.Background(), aws.String("private"), aws.String("ALIAS-srv"), &endpoint.Endpoint{
Labels: map[string]string{
endpoint.AWSSDDescriptionLabel: "ALIAS-srv",
},
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{
RoutingPolicy: aws.String(sd.RoutingPolicyWeighted),
DnsRecords: []*sd.DnsRecord{{
Type: aws.String(sd.RecordTypeA),
expectedServices["ALIAS-srv"] = &sdtypes.Service{
Name: aws.String("ALIAS-srv"),
Description: aws.String("ALIAS-srv"),
DnsConfig: &sdtypes.DnsConfig{
RoutingPolicy: sdtypes.RoutingPolicyWeighted,
DnsRecords: []sdtypes.DnsRecord{{
Type: sdtypes.RecordTypeA,
TTL: aws.Int64(100),
}},
},
@ -569,7 +604,7 @@ func TestAWSSDProvider_CreateService(t *testing.T) {
validateAWSSDServicesMapsEqual(t, expectedServices, api.services["private"])
}
func validateAWSSDServicesMapsEqual(t *testing.T, expected map[string]*sd.Service, services map[string]*sd.Service) {
func validateAWSSDServicesMapsEqual(t *testing.T, expected map[string]*sdtypes.Service, services map[string]*sdtypes.Service) {
require.Len(t, services, len(expected))
for _, srv := range services {
@ -577,31 +612,31 @@ func validateAWSSDServicesMapsEqual(t *testing.T, expected map[string]*sd.Servic
}
}
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))
func validateAWSSDServicesEqual(t *testing.T, expected *sdtypes.Service, srv *sdtypes.Service) {
assert.Equal(t, *expected.Description, *srv.Description)
assert.Equal(t, *expected.Name, *srv.Name)
assert.True(t, reflect.DeepEqual(*expected.DnsConfig, *srv.DnsConfig))
}
func TestAWSSDProvider_UpdateService(t *testing.T) {
namespaces := map[string]*sd.Namespace{
namespaces := map[string]*sdtypes.Namespace{
"private": {
Id: aws.String("private"),
Name: aws.String("private.com"),
Type: aws.String(sd.NamespaceTypeDnsPrivate),
Type: sdtypes.NamespaceTypeDnsPrivate,
},
}
services := map[string]map[string]*sd.Service{
services := map[string]map[string]*sdtypes.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),
Id: aws.String("srv1"),
Name: aws.String("service1"),
NamespaceId: aws.String("private"),
DnsConfig: &sdtypes.DnsConfig{
RoutingPolicy: sdtypes.RoutingPolicyMultivalue,
DnsRecords: []sdtypes.DnsRecord{{
Type: sdtypes.RecordTypeA,
TTL: aws.Int64(60),
}},
},
@ -617,7 +652,7 @@ func TestAWSSDProvider_UpdateService(t *testing.T) {
provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
// update service with different TTL
provider.UpdateService(services["private"]["srv1"], &endpoint.Endpoint{
provider.UpdateService(context.Background(), services["private"]["srv1"], &endpoint.Endpoint{
RecordType: endpoint.RecordTypeA,
RecordTTL: 100,
})
@ -626,15 +661,15 @@ func TestAWSSDProvider_UpdateService(t *testing.T) {
}
func TestAWSSDProvider_DeleteService(t *testing.T) {
namespaces := map[string]*sd.Namespace{
namespaces := map[string]*sdtypes.Namespace{
"private": {
Id: aws.String("private"),
Name: aws.String("private.com"),
Type: aws.String(sd.NamespaceTypeDnsPrivate),
Type: sdtypes.NamespaceTypeDnsPrivate,
},
}
services := map[string]map[string]*sd.Service{
services := map[string]map[string]*sdtypes.Service{
"private": {
"srv1": {
Id: aws.String("srv1"),
@ -665,16 +700,16 @@ func TestAWSSDProvider_DeleteService(t *testing.T) {
provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "owner-id")
// delete first service
err := provider.DeleteService(services["private"]["srv1"])
err := provider.DeleteService(context.Background(), services["private"]["srv1"])
assert.NoError(t, err)
assert.Len(t, api.services["private"], 2)
// delete third service
err1 := provider.DeleteService(services["private"]["srv3"])
err1 := provider.DeleteService(context.Background(), services["private"]["srv3"])
assert.NoError(t, err1)
assert.Len(t, api.services["private"], 1)
expectedServices := map[string]*sd.Service{
expectedServices := map[string]*sdtypes.Service{
"srv2": {
Id: aws.String("srv2"),
Description: aws.String("heritage=external-dns,external-dns/owner=owner-id"),
@ -687,48 +722,48 @@ func TestAWSSDProvider_DeleteService(t *testing.T) {
}
func TestAWSSDProvider_RegisterInstance(t *testing.T) {
namespaces := map[string]*sd.Namespace{
namespaces := map[string]*sdtypes.Namespace{
"private": {
Id: aws.String("private"),
Name: aws.String("private.com"),
Type: aws.String(sd.NamespaceTypeDnsPrivate),
Type: sdtypes.NamespaceTypeDnsPrivate,
},
}
services := map[string]map[string]*sd.Service{
services := map[string]map[string]*sdtypes.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),
Id: aws.String("a-srv"),
Name: aws.String("service1"),
NamespaceId: aws.String("private"),
DnsConfig: &sdtypes.DnsConfig{
RoutingPolicy: sdtypes.RoutingPolicyWeighted,
DnsRecords: []sdtypes.DnsRecord{{
Type: sdtypes.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),
Id: aws.String("cname-srv"),
Name: aws.String("service2"),
NamespaceId: aws.String("private"),
DnsConfig: &sdtypes.DnsConfig{
RoutingPolicy: sdtypes.RoutingPolicyWeighted,
DnsRecords: []sdtypes.DnsRecord{{
Type: sdtypes.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),
Id: aws.String("alias-srv"),
Name: aws.String("service3"),
NamespaceId: aws.String("private"),
DnsConfig: &sdtypes.DnsConfig{
RoutingPolicy: sdtypes.RoutingPolicyWeighted,
DnsRecords: []sdtypes.DnsRecord{{
Type: sdtypes.RecordTypeA,
TTL: aws.Int64(60),
}},
},
@ -739,78 +774,78 @@ func TestAWSSDProvider_RegisterInstance(t *testing.T) {
api := &AWSSDClientStub{
namespaces: namespaces,
services: services,
instances: make(map[string]map[string]*sd.Instance),
instances: make(map[string]map[string]*sdtypes.Instance),
}
provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
expectedInstances := make(map[string]*sd.Instance)
expectedInstances := make(map[string]*sdtypes.Instance)
// IP-based instance
provider.RegisterInstance(services["private"]["a-srv"], &endpoint.Endpoint{
provider.RegisterInstance(context.Background(), 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{
expectedInstances["1.2.3.4"] = &sdtypes.Instance{
Id: aws.String("1.2.3.4"),
Attributes: map[string]*string{
sdInstanceAttrIPV4: aws.String("1.2.3.4"),
Attributes: map[string]string{
sdInstanceAttrIPV4: "1.2.3.4",
},
}
expectedInstances["1.2.3.5"] = &sd.Instance{
expectedInstances["1.2.3.5"] = &sdtypes.Instance{
Id: aws.String("1.2.3.5"),
Attributes: map[string]*string{
sdInstanceAttrIPV4: aws.String("1.2.3.5"),
Attributes: map[string]string{
sdInstanceAttrIPV4: "1.2.3.5",
},
}
// AWS ELB instance (ALIAS)
provider.RegisterInstance(services["private"]["alias-srv"], &endpoint.Endpoint{
provider.RegisterInstance(context.Background(), 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{
expectedInstances["load-balancer.us-east-1.elb.amazonaws.com"] = &sdtypes.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"),
Attributes: map[string]string{
sdInstanceAttrAlias: "load-balancer.us-east-1.elb.amazonaws.com",
},
}
expectedInstances["load-balancer.us-west-2.elb.amazonaws.com"] = &sd.Instance{
expectedInstances["load-balancer.us-west-2.elb.amazonaws.com"] = &sdtypes.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"),
Attributes: map[string]string{
sdInstanceAttrAlias: "load-balancer.us-west-2.elb.amazonaws.com",
},
}
// AWS NLB instance (ALIAS)
provider.RegisterInstance(services["private"]["alias-srv"], &endpoint.Endpoint{
provider.RegisterInstance(context.Background(), services["private"]["alias-srv"], &endpoint.Endpoint{
RecordType: endpoint.RecordTypeCNAME,
DNSName: "service1.private.com.",
RecordTTL: 300,
Targets: endpoint.Targets{"load-balancer.elb.us-west-2.amazonaws.com"},
})
expectedInstances["load-balancer.elb.us-west-2.amazonaws.com"] = &sd.Instance{
expectedInstances["load-balancer.elb.us-west-2.amazonaws.com"] = &sdtypes.Instance{
Id: aws.String("load-balancer.elb.us-west-2.amazonaws.com"),
Attributes: map[string]*string{
sdInstanceAttrAlias: aws.String("load-balancer.elb.us-west-2.amazonaws.com"),
Attributes: map[string]string{
sdInstanceAttrAlias: "load-balancer.elb.us-west-2.amazonaws.com",
},
}
// CNAME instance
provider.RegisterInstance(services["private"]["cname-srv"], &endpoint.Endpoint{
provider.RegisterInstance(context.Background(), 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{
expectedInstances["cname.target.com"] = &sdtypes.Instance{
Id: aws.String("cname.target.com"),
Attributes: map[string]*string{
sdInstanceAttrCname: aws.String("cname.target.com"),
Attributes: map[string]string{
sdInstanceAttrCname: "cname.target.com",
},
}
@ -825,15 +860,15 @@ func TestAWSSDProvider_RegisterInstance(t *testing.T) {
}
func TestAWSSDProvider_DeregisterInstance(t *testing.T) {
namespaces := map[string]*sd.Namespace{
namespaces := map[string]*sdtypes.Namespace{
"private": {
Id: aws.String("private"),
Name: aws.String("private.com"),
Type: aws.String(sd.NamespaceTypeDnsPrivate),
Type: sdtypes.NamespaceTypeDnsPrivate,
},
}
services := map[string]map[string]*sd.Service{
services := map[string]map[string]*sdtypes.Service{
"private": {
"srv1": {
Id: aws.String("srv1"),
@ -842,12 +877,12 @@ func TestAWSSDProvider_DeregisterInstance(t *testing.T) {
},
}
instances := map[string]map[string]*sd.Instance{
instances := map[string]map[string]*sdtypes.Instance{
"srv1": {
"1.2.3.4": {
Id: aws.String("1.2.3.4"),
Attributes: map[string]*string{
sdInstanceAttrIPV4: aws.String("1.2.3.4"),
Attributes: map[string]string{
sdInstanceAttrIPV4: "1.2.3.4",
},
},
},
@ -861,7 +896,7 @@ func TestAWSSDProvider_DeregisterInstance(t *testing.T) {
provider := newTestAWSSDProvider(api, endpoint.NewDomainFilter([]string{}), "", "")
provider.DeregisterInstance(services["private"]["srv1"], endpoint.NewEndpoint("srv1.private.com.", endpoint.RecordTypeA, "1.2.3.4"))
provider.DeregisterInstance(context.Background(), services["private"]["srv1"], endpoint.NewEndpoint("srv1.private.com.", endpoint.RecordTypeA, "1.2.3.4"))
assert.Len(t, instances["srv1"], 0)
}

9
provider/zone_type_filter.go
View File

@ -17,8 +17,7 @@ limitations under the License.
package provider
import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/route53"
route53types "github.com/aws/aws-sdk-go-v2/service/route53/types"
)
const (
@ -52,7 +51,7 @@ func (f ZoneTypeFilter) Match(rawZoneType interface{}) bool {
case zoneTypePrivate:
return zoneType == zoneTypePrivate
}
case *route53.HostedZone:
case route53types.HostedZone:
// If the zone has no config we assume it's a public zone since the config's field
// `PrivateZone` is false by default in go.
if zoneType.Config == nil {
@ -61,9 +60,9 @@ func (f ZoneTypeFilter) Match(rawZoneType interface{}) bool {
switch f.zoneType {
case zoneTypePublic:
return !aws.BoolValue(zoneType.Config.PrivateZone)
return !zoneType.Config.PrivateZone
case zoneTypePrivate:
return aws.BoolValue(zoneType.Config.PrivateZone)
return zoneType.Config.PrivateZone
}
}

23
provider/zone_type_filter_test.go
View File

@ -19,8 +19,7 @@ package provider
import (
"testing"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/route53"
route53types "github.com/aws/aws-sdk-go-v2/service/route53/types"
"github.com/stretchr/testify/assert"
)
@ -28,8 +27,8 @@ import (
func TestZoneTypeFilterMatch(t *testing.T) {
publicZoneStr := "public"
privateZoneStr := "private"
publicZoneAWS := &route53.HostedZone{Config: &route53.HostedZoneConfig{PrivateZone: aws.Bool(false)}}
privateZoneAWS := &route53.HostedZone{Config: &route53.HostedZoneConfig{PrivateZone: aws.Bool(true)}}
publicZoneAWS := route53types.HostedZone{Config: &route53types.HostedZoneConfig{PrivateZone: false}}
privateZoneAWS := route53types.HostedZone{Config: &route53types.HostedZoneConfig{PrivateZone: true}}
for _, tc := range []struct {
zoneTypeFilter string
@ -37,10 +36,10 @@ func TestZoneTypeFilterMatch(t *testing.T) {
zones []interface{}
}{
{
"", true, []interface{}{publicZoneStr, privateZoneStr, &route53.HostedZone{}},
"", true, []interface{}{publicZoneStr, privateZoneStr, route53types.HostedZone{}},
},
{
"public", true, []interface{}{publicZoneStr, publicZoneAWS, &route53.HostedZone{}},
"public", true, []interface{}{publicZoneStr, publicZoneAWS, route53types.HostedZone{}},
},
{
"public", false, []interface{}{privateZoneStr, privateZoneAWS},
@ -49,15 +48,17 @@ func TestZoneTypeFilterMatch(t *testing.T) {
"private", true, []interface{}{privateZoneStr, privateZoneAWS},
},
{
"private", false, []interface{}{publicZoneStr, publicZoneAWS, &route53.HostedZone{}},
"private", false, []interface{}{publicZoneStr, publicZoneAWS, route53types.HostedZone{}},
},
{
"unknown", false, []interface{}{publicZoneStr},
},
} {
zoneTypeFilter := NewZoneTypeFilter(tc.zoneTypeFilter)
for _, zone := range tc.zones {
assert.Equal(t, tc.matches, zoneTypeFilter.Match(zone))
}
t.Run(tc.zoneTypeFilter, func(t *testing.T) {
zoneTypeFilter := NewZoneTypeFilter(tc.zoneTypeFilter)
for _, zone := range tc.zones {
assert.Equal(t, tc.matches, zoneTypeFilter.Match(zone))
}
})
}
}

166
registry/dynamodb.go
View File

@ -23,9 +23,10 @@ import (
"strings"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/request"
"github.com/aws/aws-sdk-go/service/dynamodb"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue"
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
dynamodbtypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
log "github.com/sirupsen/logrus"
"k8s.io/apimachinery/pkg/util/sets"
@ -34,11 +35,11 @@ import (
"sigs.k8s.io/external-dns/provider"
)
// DynamoDBAPI is the subset of the AWS Route53 API that we actually use. Add methods as required. Signatures must match exactly.
// DynamoDBAPI is the subset of the AWS DynamoDB API that we actually use. Add methods as required. Signatures must match exactly.
type DynamoDBAPI interface {
DescribeTableWithContext(ctx aws.Context, input *dynamodb.DescribeTableInput, opts ...request.Option) (*dynamodb.DescribeTableOutput, error)
ScanPagesWithContext(ctx aws.Context, input *dynamodb.ScanInput, fn func(*dynamodb.ScanOutput, bool) bool, opts ...request.Option) error
BatchExecuteStatementWithContext(aws.Context, *dynamodb.BatchExecuteStatementInput, ...request.Option) (*dynamodb.BatchExecuteStatementOutput, error)
DescribeTable(context.Context, *dynamodb.DescribeTableInput, ...func(*dynamodb.Options)) (*dynamodb.DescribeTableOutput, error)
Scan(context.Context, *dynamodb.ScanInput, ...func(*dynamodb.Options)) (*dynamodb.ScanOutput, error)
BatchExecuteStatement(context.Context, *dynamodb.BatchExecuteStatementInput, ...func(*dynamodb.Options)) (*dynamodb.BatchExecuteStatementOutput, error)
}
// DynamoDBRegistry implements registry interface with ownership implemented via an AWS DynamoDB table.
@ -225,7 +226,7 @@ func (im *DynamoDBRegistry) ApplyChanges(ctx context.Context, changes *plan.Chan
Delete: endpoint.FilterEndpointsByOwnerID(im.ownerID, changes.Delete),
}
statements := make([]*dynamodb.BatchStatementRequest, 0, len(filteredChanges.Create)+len(filteredChanges.UpdateNew))
statements := make([]dynamodbtypes.BatchStatementRequest, 0, len(filteredChanges.Create)+len(filteredChanges.UpdateNew))
for _, r := range filteredChanges.Create {
if r.Labels == nil {
r.Labels = make(map[string]string)
@ -286,12 +287,15 @@ func (im *DynamoDBRegistry) ApplyChanges(ctx context.Context, changes *plan.Chan
}
}
err := im.executeStatements(ctx, statements, func(request *dynamodb.BatchStatementRequest, response *dynamodb.BatchStatementResponse) error {
err := im.executeStatements(ctx, statements, func(request dynamodbtypes.BatchStatementRequest, response dynamodbtypes.BatchStatementResponse) error {
var context string
if strings.HasPrefix(*request.Statement, "INSERT") {
if aws.StringValue(response.Error.Code) == "DuplicateItem" {
if response.Error.Code == dynamodbtypes.BatchStatementErrorCodeEnumDuplicateItem {
// We lost a race with a different owner or another owner has an orphaned ownership record.
key := fromDynamoKey(request.Parameters[0])
key, err := fromDynamoKey(request.Parameters[0])
if err != nil {
return err
}
for i, endpoint := range filteredChanges.Create {
if endpoint.Key() == key {
log.Infof("Skipping endpoint %v because owner does not match", endpoint)
@ -303,11 +307,19 @@ func (im *DynamoDBRegistry) ApplyChanges(ctx context.Context, changes *plan.Chan
}
}
}
context = fmt.Sprintf("inserting dynamodb record %q", aws.StringValue(request.Parameters[0].S))
var record string
if err := attributevalue.Unmarshal(request.Parameters[0], &record); err != nil {
return fmt.Errorf("inserting dynamodb record: %w", err)
}
context = fmt.Sprintf("inserting dynamodb record %q", record)
} else {
context = fmt.Sprintf("updating dynamodb record %q", aws.StringValue(request.Parameters[1].S))
var record string
if err := attributevalue.Unmarshal(request.Parameters[1], &record); err != nil {
return fmt.Errorf("inserting dynamodb record: %w", err)
}
context = fmt.Sprintf("updating dynamodb record %q", record)
}
return fmt.Errorf("%s: %s: %s", context, aws.StringValue(response.Error.Code), aws.StringValue(response.Error.Message))
return fmt.Errorf("%s: %s: %s", context, response.Error.Code, *response.Error.Message)
})
if err != nil {
im.recordsCache = nil
@ -326,7 +338,7 @@ func (im *DynamoDBRegistry) ApplyChanges(ctx context.Context, changes *plan.Chan
return err
}
statements = make([]*dynamodb.BatchStatementRequest, 0, len(filteredChanges.Delete)+len(im.orphanedLabels))
statements = make([]dynamodbtypes.BatchStatementRequest, 0, len(filteredChanges.Delete)+len(im.orphanedLabels))
for _, r := range filteredChanges.Delete {
statements = im.appendDelete(statements, r.Key())
}
@ -335,9 +347,13 @@ func (im *DynamoDBRegistry) ApplyChanges(ctx context.Context, changes *plan.Chan
delete(im.labels, r)
}
im.orphanedLabels = nil
return im.executeStatements(ctx, statements, func(request *dynamodb.BatchStatementRequest, response *dynamodb.BatchStatementResponse) error {
return im.executeStatements(ctx, statements, func(request dynamodbtypes.BatchStatementRequest, response dynamodbtypes.BatchStatementResponse) error {
im.labels = nil
return fmt.Errorf("deleting dynamodb record %q: %s: %s", aws.StringValue(request.Parameters[0].S), aws.StringValue(response.Error.Code), aws.StringValue(response.Error.Message))
record, err := fromDynamoKey(request.Parameters[0])
if err != nil {
return fmt.Errorf("deleting dynamodb record: %w", err)
}
return fmt.Errorf("deleting dynamodb record %q: %s: %s", record, response.Error.Code, *response.Error.Message)
})
}
@ -347,7 +363,7 @@ func (im *DynamoDBRegistry) AdjustEndpoints(endpoints []*endpoint.Endpoint) ([]*
}
func (im *DynamoDBRegistry) readLabels(ctx context.Context) error {
table, err := im.dynamodbAPI.DescribeTableWithContext(ctx, &dynamodb.DescribeTableInput{
table, err := im.dynamodbAPI.DescribeTable(ctx, &dynamodb.DescribeTableInput{
TableName: aws.String(im.table),
})
if err != nil {
@ -356,8 +372,8 @@ func (im *DynamoDBRegistry) readLabels(ctx context.Context) error {
foundKey := false
for _, def := range table.Table.AttributeDefinitions {
if aws.StringValue(def.AttributeName) == "k" {
if aws.StringValue(def.AttributeType) != "S" {
if *def.AttributeName == "k" {
if def.AttributeType != dynamodbtypes.ScalarAttributeTypeS {
return fmt.Errorf("table %q attribute \"k\" must have type \"S\"", im.table)
}
foundKey = true
@ -367,7 +383,7 @@ func (im *DynamoDBRegistry) readLabels(ctx context.Context) error {
return fmt.Errorf("table %q must have attribute \"k\" of type \"S\"", im.table)
}
if aws.StringValue(table.Table.KeySchema[0].AttributeName) != "k" {
if *table.Table.KeySchema[0].AttributeName != "k" {
return fmt.Errorf("table %q must have hash key \"k\"", im.table)
}
if len(table.Table.KeySchema) > 1 {
@ -375,76 +391,92 @@ func (im *DynamoDBRegistry) readLabels(ctx context.Context) error {
}
labels := map[endpoint.EndpointKey]endpoint.Labels{}
err = im.dynamodbAPI.ScanPagesWithContext(ctx, &dynamodb.ScanInput{
scanPaginator := dynamodb.NewScanPaginator(im.dynamodbAPI, &dynamodb.ScanInput{
TableName: aws.String(im.table),
FilterExpression: aws.String("o = :ownerval"),
ExpressionAttributeValues: map[string]*dynamodb.AttributeValue{
":ownerval": {S: aws.String(im.ownerID)},
ExpressionAttributeValues: map[string]dynamodbtypes.AttributeValue{
":ownerval": &dynamodbtypes.AttributeValueMemberS{Value: im.ownerID},
},
ProjectionExpression: aws.String("k,l"),
ConsistentRead: aws.Bool(true),
}, func(output *dynamodb.ScanOutput, last bool) bool {
for _, item := range output.Items {
labels[fromDynamoKey(item["k"])] = fromDynamoLabels(item["l"], im.ownerID)
}
return true
})
if err != nil {
return fmt.Errorf("querying dynamodb: %w", err)
for scanPaginator.HasMorePages() {
output, err := scanPaginator.NextPage(ctx)
if err != nil {
return fmt.Errorf("scanning table %q: %w", im.table, err)
}
for _, item := range output.Items {
k, err := fromDynamoKey(item["k"])
if err != nil {
return fmt.Errorf("querying dynamodb for key: %w", err)
}
l, err := fromDynamoLabels(item["l"], im.ownerID)
if err != nil {
return fmt.Errorf("querying dynamodb for labels: %w", err)
}
labels[k] = l
}
}
im.labels = labels
return nil
}
func fromDynamoKey(key *dynamodb.AttributeValue) endpoint.EndpointKey {
split := strings.SplitN(aws.StringValue(key.S), "#", 3)
func fromDynamoKey(key dynamodbtypes.AttributeValue) (endpoint.EndpointKey, error) {
var ep string
if err := attributevalue.Unmarshal(key, &ep); err != nil {
return endpoint.EndpointKey{}, fmt.Errorf("unmarshalling endpoint key: %w", err)
}
split := strings.SplitN(ep, "#", 3)
return endpoint.EndpointKey{
DNSName: split[0],
RecordType: split[1],
SetIdentifier: split[2],
}, nil
}
func toDynamoKey(key endpoint.EndpointKey) dynamodbtypes.AttributeValue {
return &dynamodbtypes.AttributeValueMemberS{
Value: fmt.Sprintf("%s#%s#%s", key.DNSName, key.RecordType, key.SetIdentifier),
}
}
func toDynamoKey(key endpoint.EndpointKey) *dynamodb.AttributeValue {
return &dynamodb.AttributeValue{
S: aws.String(fmt.Sprintf("%s#%s#%s", key.DNSName, key.RecordType, key.SetIdentifier)),
}
}
func fromDynamoLabels(label *dynamodb.AttributeValue, owner string) endpoint.Labels {
func fromDynamoLabels(label dynamodbtypes.AttributeValue, owner string) (endpoint.Labels, error) {
labels := endpoint.NewLabels()
for k, v := range label.M {
labels[k] = aws.StringValue(v.S)
if err := attributevalue.Unmarshal(label, &labels); err != nil {
return endpoint.Labels{}, fmt.Errorf("unmarshalling labels: %w", err)
}
labels[endpoint.OwnerLabelKey] = owner
return labels
return labels, nil
}
func toDynamoLabels(labels endpoint.Labels) *dynamodb.AttributeValue {
labelMap := make(map[string]*dynamodb.AttributeValue, len(labels))
func toDynamoLabels(labels endpoint.Labels) dynamodbtypes.AttributeValue {
labelMap := make(map[string]dynamodbtypes.AttributeValue, len(labels))
for k, v := range labels {
if k == endpoint.OwnerLabelKey {
continue
}
labelMap[k] = &dynamodb.AttributeValue{S: aws.String(v)}
labelMap[k] = &dynamodbtypes.AttributeValueMemberS{Value: v}
}
return &dynamodb.AttributeValue{M: labelMap}
return &dynamodbtypes.AttributeValueMemberM{Value: labelMap}
}
func (im *DynamoDBRegistry) appendInsert(statements []*dynamodb.BatchStatementRequest, key endpoint.EndpointKey, new endpoint.Labels) []*dynamodb.BatchStatementRequest {
return append(statements, &dynamodb.BatchStatementRequest{
Statement: aws.String(fmt.Sprintf("INSERT INTO %q VALUE {'k':?, 'o':?, 'l':?}", im.table)),
Parameters: []*dynamodb.AttributeValue{
func (im *DynamoDBRegistry) appendInsert(statements []dynamodbtypes.BatchStatementRequest, key endpoint.EndpointKey, new endpoint.Labels) []dynamodbtypes.BatchStatementRequest {
return append(statements, dynamodbtypes.BatchStatementRequest{
Statement: aws.String(fmt.Sprintf("INSERT INTO %q VALUE {'k':?, 'o':?, 'l':?}", im.table)),
ConsistentRead: aws.Bool(true),
Parameters: []dynamodbtypes.AttributeValue{
toDynamoKey(key),
{S: aws.String(im.ownerID)},
&dynamodbtypes.AttributeValueMemberS{
Value: im.ownerID,
},
toDynamoLabels(new),
},
ConsistentRead: aws.Bool(true),
})
}
func (im *DynamoDBRegistry) appendUpdate(statements []*dynamodb.BatchStatementRequest, key endpoint.EndpointKey, old endpoint.Labels, new endpoint.Labels) []*dynamodb.BatchStatementRequest {
func (im *DynamoDBRegistry) appendUpdate(statements []dynamodbtypes.BatchStatementRequest, key endpoint.EndpointKey, old endpoint.Labels, new endpoint.Labels) []dynamodbtypes.BatchStatementRequest {
if len(old) == len(new) {
equal := true
for k, v := range old {
@ -458,28 +490,28 @@ func (im *DynamoDBRegistry) appendUpdate(statements []*dynamodb.BatchStatementRe
}
}
return append(statements, &dynamodb.BatchStatementRequest{
return append(statements, dynamodbtypes.BatchStatementRequest{
Statement: aws.String(fmt.Sprintf("UPDATE %q SET \"l\"=? WHERE \"k\"=?", im.table)),
Parameters: []*dynamodb.AttributeValue{
Parameters: []dynamodbtypes.AttributeValue{
toDynamoLabels(new),
toDynamoKey(key),
},
})
}
func (im *DynamoDBRegistry) appendDelete(statements []*dynamodb.BatchStatementRequest, key endpoint.EndpointKey) []*dynamodb.BatchStatementRequest {
return append(statements, &dynamodb.BatchStatementRequest{
func (im *DynamoDBRegistry) appendDelete(statements []dynamodbtypes.BatchStatementRequest, key endpoint.EndpointKey) []dynamodbtypes.BatchStatementRequest {
return append(statements, dynamodbtypes.BatchStatementRequest{
Statement: aws.String(fmt.Sprintf("DELETE FROM %q WHERE \"k\"=? AND \"o\"=?", im.table)),
Parameters: []*dynamodb.AttributeValue{
Parameters: []dynamodbtypes.AttributeValue{
toDynamoKey(key),
{S: aws.String(im.ownerID)},
&dynamodbtypes.AttributeValueMemberS{Value: im.ownerID},
},
})
}
func (im *DynamoDBRegistry) executeStatements(ctx context.Context, statements []*dynamodb.BatchStatementRequest, handleErr func(request *dynamodb.BatchStatementRequest, response *dynamodb.BatchStatementResponse) error) error {
func (im *DynamoDBRegistry) executeStatements(ctx context.Context, statements []dynamodbtypes.BatchStatementRequest, handleErr func(request dynamodbtypes.BatchStatementRequest, response dynamodbtypes.BatchStatementResponse) error) error {
for len(statements) > 0 {
var chunk []*dynamodb.BatchStatementRequest
var chunk []dynamodbtypes.BatchStatementRequest
if len(statements) > int(dynamodbMaxBatchSize) {
chunk = statements[:dynamodbMaxBatchSize]
statements = statements[dynamodbMaxBatchSize:]
@ -488,7 +520,7 @@ func (im *DynamoDBRegistry) executeStatements(ctx context.Context, statements []
statements = nil
}
output, err := im.dynamodbAPI.BatchExecuteStatementWithContext(ctx, &dynamodb.BatchExecuteStatementInput{
output, err := im.dynamodbAPI.BatchExecuteStatement(ctx, &dynamodb.BatchExecuteStatementInput{
Statements: chunk,
})
if err != nil {
@ -501,9 +533,13 @@ func (im *DynamoDBRegistry) executeStatements(ctx context.Context, statements []
op, _, _ := strings.Cut(*request.Statement, " ")
var key string
if op == "UPDATE" {
key = *request.Parameters[1].S
if err := attributevalue.Unmarshal(request.Parameters[1], &key); err != nil {
return err
}
} else {
key = *request.Parameters[0].S
if err := attributevalue.Unmarshal(request.Parameters[0], &key); err != nil {
return err
}
}
log.Infof("%s dynamodb record %q", op, key)
} else {

150
registry/dynamodb_test.go
View File

@ -22,9 +22,10 @@ import (
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/request"
"github.com/aws/aws-sdk-go/service/dynamodb"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue"
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
dynamodbtypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"k8s.io/apimachinery/pkg/util/sets"
@ -69,40 +70,40 @@ func TestDynamoDBRegistryNew(t *testing.T) {
func TestDynamoDBRegistryRecordsBadTable(t *testing.T) {
for _, tc := range []struct {
name string
setup func(desc *dynamodb.TableDescription)
setup func(desc *dynamodbtypes.TableDescription)
expected string
}{
{
name: "missing attribute k",
setup: func(desc *dynamodb.TableDescription) {
setup: func(desc *dynamodbtypes.TableDescription) {
desc.AttributeDefinitions[0].AttributeName = aws.String("wrong")
},
expected: "table \"test-table\" must have attribute \"k\" of type \"S\"",
},
{
name: "wrong attribute type",
setup: func(desc *dynamodb.TableDescription) {
desc.AttributeDefinitions[0].AttributeType = aws.String("SS")
setup: func(desc *dynamodbtypes.TableDescription) {
desc.AttributeDefinitions[0].AttributeType = "SS"
},
expected: "table \"test-table\" attribute \"k\" must have type \"S\"",
},
{
name: "wrong key",
setup: func(desc *dynamodb.TableDescription) {
setup: func(desc *dynamodbtypes.TableDescription) {
desc.KeySchema[0].AttributeName = aws.String("wrong")
},
expected: "table \"test-table\" must have hash key \"k\"",
},
{
name: "has range key",
setup: func(desc *dynamodb.TableDescription) {
desc.AttributeDefinitions = append(desc.AttributeDefinitions, &dynamodb.AttributeDefinition{
setup: func(desc *dynamodbtypes.TableDescription) {
desc.AttributeDefinitions = append(desc.AttributeDefinitions, dynamodbtypes.AttributeDefinition{
AttributeName: aws.String("o"),
AttributeType: aws.String("S"),
AttributeType: dynamodbtypes.ScalarAttributeTypeS,
})
desc.KeySchema = append(desc.KeySchema, &dynamodb.KeySchemaElement{
desc.KeySchema = append(desc.KeySchema, dynamodbtypes.KeySchemaElement{
AttributeName: aws.String("o"),
KeyType: aws.String("RANGE"),
KeyType: dynamodbtypes.KeyTypeRange,
})
},
expected: "table \"test-table\" must not have a range key",
@ -559,8 +560,8 @@ func TestDynamoDBRegistryApplyChanges(t *testing.T) {
},
},
stubConfig: DynamoDBStubConfig{
ExpectInsertError: map[string]string{
"new.test-zone.example.org#CNAME#set-new": "DuplicateItem",
ExpectInsertError: map[string]dynamodbtypes.BatchStatementErrorCodeEnum{
"new.test-zone.example.org#CNAME#set-new": dynamodbtypes.BatchStatementErrorCodeEnumDuplicateItem,
},
ExpectDelete: sets.New("quux.test-zone.example.org#A#set-2"),
},
@ -620,7 +621,7 @@ func TestDynamoDBRegistryApplyChanges(t *testing.T) {
},
},
stubConfig: DynamoDBStubConfig{
ExpectInsertError: map[string]string{
ExpectInsertError: map[string]dynamodbtypes.BatchStatementErrorCodeEnum{
"new.test-zone.example.org#CNAME#set-new": "TestingError",
},
},
@ -928,7 +929,7 @@ func TestDynamoDBRegistryApplyChanges(t *testing.T) {
},
},
stubConfig: DynamoDBStubConfig{
ExpectUpdateError: map[string]string{
ExpectUpdateError: map[string]dynamodbtypes.BatchStatementErrorCodeEnum{
"bar.test-zone.example.org#CNAME#": "TestingError",
},
},
@ -1073,15 +1074,15 @@ func TestDynamoDBRegistryApplyChanges(t *testing.T) {
type DynamoDBStub struct {
t *testing.T
stubConfig *DynamoDBStubConfig
tableDescription dynamodb.TableDescription
tableDescription dynamodbtypes.TableDescription
changesApplied bool
}
type DynamoDBStubConfig struct {
ExpectInsert map[string]map[string]string
ExpectInsertError map[string]string
ExpectInsertError map[string]dynamodbtypes.BatchStatementErrorCodeEnum
ExpectUpdate map[string]map[string]string
ExpectUpdateError map[string]string
ExpectUpdateError map[string]dynamodbtypes.BatchStatementErrorCodeEnum
ExpectDelete sets.Set[string]
}
@ -1100,17 +1101,17 @@ func newDynamoDBAPIStub(t *testing.T, stubConfig *DynamoDBStubConfig) (*DynamoDB
stub := &DynamoDBStub{
t: t,
stubConfig: stubConfig,
tableDescription: dynamodb.TableDescription{
AttributeDefinitions: []*dynamodb.AttributeDefinition{
tableDescription: dynamodbtypes.TableDescription{
AttributeDefinitions: []dynamodbtypes.AttributeDefinition{
{
AttributeName: aws.String("k"),
AttributeType: aws.String("S"),
AttributeType: dynamodbtypes.ScalarAttributeTypeS,
},
},
KeySchema: []*dynamodb.KeySchemaElement{
KeySchema: []dynamodbtypes.KeySchemaElement{
{
AttributeName: aws.String("k"),
KeyType: aws.String("HASH"),
KeyType: dynamodbtypes.KeyTypeHash,
},
},
},
@ -1131,7 +1132,7 @@ func newDynamoDBAPIStub(t *testing.T, stubConfig *DynamoDBStubConfig) (*DynamoDB
}
}
func (r *DynamoDBStub) DescribeTableWithContext(ctx aws.Context, input *dynamodb.DescribeTableInput, opts ...request.Option) (*dynamodb.DescribeTableOutput, error) {
func (r *DynamoDBStub) DescribeTable(ctx context.Context, input *dynamodb.DescribeTableInput, opts ...func(*dynamodb.Options)) (*dynamodb.DescribeTableOutput, error) {
assert.NotNil(r.t, ctx)
assert.Equal(r.t, "test-table", *input.TableName, "table name")
return &dynamodb.DescribeTableOutput{
@ -1139,75 +1140,80 @@ func (r *DynamoDBStub) DescribeTableWithContext(ctx aws.Context, input *dynamodb
}, nil
}
func (r *DynamoDBStub) ScanPagesWithContext(ctx aws.Context, input *dynamodb.ScanInput, fn func(*dynamodb.ScanOutput, bool) bool, opts ...request.Option) error {
func (r *DynamoDBStub) Scan(ctx context.Context, input *dynamodb.ScanInput, opts ...func(*dynamodb.Options)) (*dynamodb.ScanOutput, error) {
assert.NotNil(r.t, ctx)
assert.Equal(r.t, "test-table", *input.TableName, "table name")
assert.Equal(r.t, "o = :ownerval", *input.FilterExpression)
assert.Len(r.t, input.ExpressionAttributeValues, 1)
assert.Equal(r.t, "test-owner", *input.ExpressionAttributeValues[":ownerval"].S)
var owner string
assert.Nil(r.t, attributevalue.Unmarshal(input.ExpressionAttributeValues[":ownerval"], &owner))
assert.Equal(r.t, "test-owner", owner)
assert.Equal(r.t, "k,l", *input.ProjectionExpression)
assert.True(r.t, *input.ConsistentRead)
fn(&dynamodb.ScanOutput{
Items: []map[string]*dynamodb.AttributeValue{
return &dynamodb.ScanOutput{
Items: []map[string]dynamodbtypes.AttributeValue{
{
"k": &dynamodb.AttributeValue{S: aws.String("bar.test-zone.example.org#CNAME#")},
"l": &dynamodb.AttributeValue{M: map[string]*dynamodb.AttributeValue{
endpoint.ResourceLabelKey: {S: aws.String("ingress/default/my-ingress")},
"k": &dynamodbtypes.AttributeValueMemberS{Value: "bar.test-zone.example.org#CNAME#"},
"l": &dynamodbtypes.AttributeValueMemberM{Value: map[string]dynamodbtypes.AttributeValue{
endpoint.ResourceLabelKey: &dynamodbtypes.AttributeValueMemberS{Value: "ingress/default/my-ingress"},
}},
},
{
"k": &dynamodb.AttributeValue{S: aws.String("baz.test-zone.example.org#A#set-1")},
"l": &dynamodb.AttributeValue{M: map[string]*dynamodb.AttributeValue{
endpoint.ResourceLabelKey: {S: aws.String("ingress/default/my-ingress")},
"k": &dynamodbtypes.AttributeValueMemberS{Value: "baz.test-zone.example.org#A#set-1"},
"l": &dynamodbtypes.AttributeValueMemberM{Value: map[string]dynamodbtypes.AttributeValue{
endpoint.ResourceLabelKey: &dynamodbtypes.AttributeValueMemberS{Value: "ingress/default/my-ingress"},
}},
},
{
"k": &dynamodb.AttributeValue{S: aws.String("baz.test-zone.example.org#A#set-2")},
"l": &dynamodb.AttributeValue{M: map[string]*dynamodb.AttributeValue{
endpoint.ResourceLabelKey: {S: aws.String("ingress/default/other-ingress")},
"k": &dynamodbtypes.AttributeValueMemberS{Value: "baz.test-zone.example.org#A#set-2"},
"l": &dynamodbtypes.AttributeValueMemberM{Value: map[string]dynamodbtypes.AttributeValue{
endpoint.ResourceLabelKey: &dynamodbtypes.AttributeValueMemberS{Value: "ingress/default/other-ingress"},
}},
},
{
"k": &dynamodb.AttributeValue{S: aws.String("quux.test-zone.example.org#A#set-2")},
"l": &dynamodb.AttributeValue{M: map[string]*dynamodb.AttributeValue{
endpoint.ResourceLabelKey: {S: aws.String("ingress/default/quux-ingress")},
"k": &dynamodbtypes.AttributeValueMemberS{Value: "quux.test-zone.example.org#A#set-2"},
"l": &dynamodbtypes.AttributeValueMemberM{Value: map[string]dynamodbtypes.AttributeValue{
endpoint.ResourceLabelKey: &dynamodbtypes.AttributeValueMemberS{Value: "ingress/default/quux-ingress"},
}},
},
},
}, true)
return nil
}, nil
}
func (r *DynamoDBStub) BatchExecuteStatementWithContext(context aws.Context, input *dynamodb.BatchExecuteStatementInput, option ...request.Option) (*dynamodb.BatchExecuteStatementOutput, error) {
func (r *DynamoDBStub) BatchExecuteStatement(context context.Context, input *dynamodb.BatchExecuteStatementInput, option ...func(*dynamodb.Options)) (*dynamodb.BatchExecuteStatementOutput, error) {
assert.NotNil(r.t, context)
hasDelete := strings.HasPrefix(strings.ToLower(aws.StringValue(input.Statements[0].Statement)), "delete")
hasDelete := strings.HasPrefix(strings.ToLower(*input.Statements[0].Statement), "delete")
assert.Equal(r.t, hasDelete, r.changesApplied, "delete after provider changes, everything else before")
assert.LessOrEqual(r.t, len(input.Statements), 25)
responses := make([]*dynamodb.BatchStatementResponse, 0, len(input.Statements))
responses := make([]dynamodbtypes.BatchStatementResponse, 0, len(input.Statements))
for _, statement := range input.Statements {
assert.Equal(r.t, hasDelete, strings.HasPrefix(strings.ToLower(aws.StringValue(statement.Statement)), "delete"))
switch aws.StringValue(statement.Statement) {
assert.Equal(r.t, hasDelete, strings.HasPrefix(strings.ToLower(*statement.Statement), "delete"))
switch *statement.Statement {
case "DELETE FROM \"test-table\" WHERE \"k\"=? AND \"o\"=?":
assert.True(r.t, r.changesApplied, "unexpected delete before provider changes")
key := aws.StringValue(statement.Parameters[0].S)
var key string
assert.Nil(r.t, attributevalue.Unmarshal(statement.Parameters[0], &key))
assert.True(r.t, r.stubConfig.ExpectDelete.Has(key), "unexpected delete for key %q", key)
r.stubConfig.ExpectDelete.Delete(key)
assert.Equal(r.t, "test-owner", aws.StringValue(statement.Parameters[1].S))
var testOwner string
assert.Nil(r.t, attributevalue.Unmarshal(statement.Parameters[1], &testOwner))
assert.Equal(r.t, "test-owner", testOwner)
responses = append(responses, &dynamodb.BatchStatementResponse{})
responses = append(responses, dynamodbtypes.BatchStatementResponse{})
case "INSERT INTO \"test-table\" VALUE {'k':?, 'o':?, 'l':?}":
assert.False(r.t, r.changesApplied, "unexpected insert after provider changes")
key := aws.StringValue(statement.Parameters[0].S)
var key string
assert.Nil(r.t, attributevalue.Unmarshal(statement.Parameters[0], &key))
if code, exists := r.stubConfig.ExpectInsertError[key]; exists {
delete(r.stubConfig.ExpectInsertError, key)
responses = append(responses, &dynamodb.BatchStatementResponse{
Error: &dynamodb.BatchStatementError{
Code: aws.String(code),
responses = append(responses, dynamodbtypes.BatchStatementResponse{
Error: &dynamodbtypes.BatchStatementError{
Code: code,
Message: aws.String("testing error"),
},
})
@ -1218,10 +1224,15 @@ func (r *DynamoDBStub) BatchExecuteStatementWithContext(context aws.Context, inp
assert.True(r.t, found, "unexpected insert for key %q", key)
delete(r.stubConfig.ExpectInsert, key)
assert.Equal(r.t, "test-owner", aws.StringValue(statement.Parameters[1].S))
var testOwner string
assert.Nil(r.t, attributevalue.Unmarshal(statement.Parameters[1], &testOwner))
assert.Equal(r.t, "test-owner", testOwner)
for label, attribute := range statement.Parameters[2].M {
value := aws.StringValue(attribute.S)
var labels map[string]string
err := attributevalue.Unmarshal(statement.Parameters[2], &labels)
assert.Nil(r.t, err)
for label, value := range labels {
expectedValue, found := expectedLabels[label]
assert.True(r.t, found, "insert for key %q has unexpected label %q", key, label)
delete(expectedLabels, label)
@ -1232,17 +1243,18 @@ func (r *DynamoDBStub) BatchExecuteStatementWithContext(context aws.Context, inp
r.t.Errorf("insert for key %q did not get expected label %q", key, label)
}
responses = append(responses, &dynamodb.BatchStatementResponse{})
responses = append(responses, dynamodbtypes.BatchStatementResponse{})
case "UPDATE \"test-table\" SET \"l\"=? WHERE \"k\"=?":
assert.False(r.t, r.changesApplied, "unexpected update after provider changes")
key := aws.StringValue(statement.Parameters[1].S)
var key string
assert.Nil(r.t, attributevalue.Unmarshal(statement.Parameters[1], &key))
if code, exists := r.stubConfig.ExpectUpdateError[key]; exists {
delete(r.stubConfig.ExpectInsertError, key)
responses = append(responses, &dynamodb.BatchStatementResponse{
Error: &dynamodb.BatchStatementError{
Code: aws.String(code),
responses = append(responses, dynamodbtypes.BatchStatementResponse{
Error: &dynamodbtypes.BatchStatementError{
Code: code,
Message: aws.String("testing error"),
},
})
@ -1253,8 +1265,10 @@ func (r *DynamoDBStub) BatchExecuteStatementWithContext(context aws.Context, inp
assert.True(r.t, found, "unexpected update for key %q", key)
delete(r.stubConfig.ExpectUpdate, key)
for label, attribute := range statement.Parameters[0].M {
value := aws.StringValue(attribute.S)
var labels map[string]string
assert.Nil(r.t, attributevalue.Unmarshal(statement.Parameters[0], &labels))
for label, value := range labels {
expectedValue, found := expectedLabels[label]
assert.True(r.t, found, "update for key %q has unexpected label %q", key, label)
delete(expectedLabels, label)
@ -1265,10 +1279,10 @@ func (r *DynamoDBStub) BatchExecuteStatementWithContext(context aws.Context, inp
r.t.Errorf("update for key %q did not get expected label %q", key, label)
}
responses = append(responses, &dynamodb.BatchStatementResponse{})
responses = append(responses, dynamodbtypes.BatchStatementResponse{})
default:
r.t.Errorf("unexpected statement: %s", aws.StringValue(statement.Statement))
r.t.Errorf("unexpected statement: %s", *statement.Statement)
}
}