/* Copyright 2018 The Kubernetes Authors. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. */ package provider import ( "fmt" "os" "strconv" "strings" "time" log "github.com/sirupsen/logrus" "github.com/nesv/go-dynect/dynect" "github.com/kubernetes-incubator/external-dns/endpoint" "github.com/kubernetes-incubator/external-dns/plan" ) const ( // 10 minutes default timeout if not configured using flags dynDefaultTTL = 600 // can store 20000 entries globally, that's about 4MB of memory // may be made configurable in the future but 20K records seems like enough for a few zones cacheMaxSize = 20000 // two consecutive bad logins happen at least this many seconds appart // While it is easy to get the username right, misconfiguring the password // can get account blocked. Exit(1) is not a good solution // as k8s will restart the pod and another login attempt will be made badLoginMinIntervalSeconds = 30 * 60 // this prefix must be stripped from resource links before feeding them to dynect.Client.Do() restAPIPrefix = "/REST/" ) // A simple non-thread-safe cache with TTL. The TTL of the records is used here to // This cache is used to save on requests to DynAPI type cache struct { contents map[string]*entry } type entry struct { expires int64 ep *endpoint.Endpoint } func (c *cache) Put(link string, ep *endpoint.Endpoint) { // flush the whole cache on overflow if len(c.contents) >= cacheMaxSize { c.contents = make(map[string]*entry) } c.contents[link] = &entry{ ep: ep, expires: unixNow() + int64(ep.RecordTTL), } } func unixNow() int64 { return int64(time.Now().Unix()) } func (c *cache) Get(link string) *endpoint.Endpoint { result, ok := c.contents[link] if !ok { return nil } now := unixNow() if result.expires < now { delete(c.contents, link) return nil } return result.ep } // DynConfig hold connection parameters to dyn.com and interanl state type DynConfig struct { DomainFilter DomainFilter ZoneIDFilter ZoneIDFilter DryRun bool CustomerName string Username string Password string MinTTLSeconds int AppVersion string DynVersion string } // DynProvider is the actual interface impl. type dynProviderState struct { DynConfig Cache *cache LastLoginErrorTime int64 } // ZoneChange is missing from dynect: https://help.dyn.com/get-zone-changeset-api/ type ZoneChange struct { ID int `json:"id"` UserID int `json:"user_id"` Zone string `json:"zone"` FQDN string `json:"FQDN"` Serial int `json:"serial"` TTL int `json:"ttl"` Type string `json:"rdata_type"` RData dynect.DataBlock `json:"rdata"` } // ZoneChangesResponse is missing from dynect: https://help.dyn.com/get-zone-changeset-api/ type ZoneChangesResponse struct { dynect.ResponseBlock Data []ZoneChange `json:"data"` } // ZonePublishRequest is missing from dynect but the notes field is a nice place to let // external-dns report some internal info during commit type ZonePublishRequest struct { Publish bool `json:"publish"` Notes string `json:"notes"` } // ZonePublisResponse holds the status after publish type ZonePublisResponse struct { dynect.ResponseBlock Data map[string]interface{} `json:"data"` } // NewDynProvider initializes a new Dyn Provider. func NewDynProvider(config DynConfig) (Provider, error) { return &dynProviderState{ DynConfig: config, Cache: &cache{ contents: make(map[string]*entry), }, }, nil } // filterAndFixLinks removes from `links` all the records we don't care about // and strops the /REST/ prefix func filterAndFixLinks(links []string, filter DomainFilter) []string { var result []string for _, link := range links { link = strings.TrimPrefix(link, restAPIPrefix) // covert resource link to just the FQDN so we can filter on it domain := link[0:strings.LastIndexByte(link, '/')] domain = domain[strings.LastIndexByte(domain, '/')+1:] // simply ignore all record types we don't care about if !strings.HasPrefix(link, endpoint.RecordTypeA) && !strings.HasPrefix(link, endpoint.RecordTypeCNAME) && !strings.HasPrefix(link, endpoint.RecordTypeTXT) { continue } if filter.Match(domain) { result = append(result, link) } } return result } func fixMissingTTL(ttl endpoint.TTL, minTTLSeconds int) string { i := dynDefaultTTL if ttl.IsConfigured() { if int(ttl) < minTTLSeconds { i = minTTLSeconds } else { i = int(ttl) } } return strconv.Itoa(i) } // merge produces a singe list of records that can be used as a replacement. // Dyn allows to replace all records with a single call // Invariant: the result contains only elements from the updateNew parameter func merge(updateOld, updateNew []*endpoint.Endpoint) []*endpoint.Endpoint { findMatch := func(template *endpoint.Endpoint) *endpoint.Endpoint { for _, new := range updateNew { if template.DNSName == new.DNSName && template.RecordType == new.RecordType { return new } } return nil } var result []*endpoint.Endpoint for _, old := range updateOld { matchingNew := findMatch(old) if matchingNew == nil { // no match, shouldn't happen continue } if !matchingNew.Targets.Same(old.Targets) { // new target: always update, TTL will be overwritten too if necessary result = append(result, matchingNew) continue } if matchingNew.RecordTTL != 0 && matchingNew.RecordTTL != old.RecordTTL { // same target, but new non-zero TTL set in k8s, must update // probably would happen only if there is a bug in the code calling the provider result = append(result, matchingNew) } } return result } // extractTarget populates the correct field given a record type. // See dynect.DataBlock comments for details. Empty response means nothing // was populated - basically an error func extractTarget(recType string, data *dynect.DataBlock) string { result := "" if recType == endpoint.RecordTypeA { result = data.Address } if recType == endpoint.RecordTypeCNAME { result = data.CName result = strings.TrimSuffix(result, ".") } if recType == endpoint.RecordTypeTXT { result = data.TxtData } return result } // recordLinkToEndpoint makes an Endpoint given a resource link optinally making a remote call if a cached entry is expired func (d *dynProviderState) recordLinkToEndpoint(client *dynect.Client, recordLink string) (*endpoint.Endpoint, error) { result := d.Cache.Get(recordLink) if result != nil { log.Infof("Using cached endpoint for %s: %+v", recordLink, result) return result, nil } rec := dynect.RecordResponse{} err := client.Do("GET", recordLink, nil, &rec) if err != nil { return nil, err } // ignore all records but the types supported by external- target := extractTarget(rec.Data.RecordType, &rec.Data.RData) if target == "" { return nil, nil } result = &endpoint.Endpoint{ DNSName: rec.Data.FQDN, RecordTTL: endpoint.TTL(rec.Data.TTL), RecordType: rec.Data.RecordType, Targets: endpoint.Targets{target}, } log.Debugf("Fetched new endpoint for %s: %+v", recordLink, result) d.Cache.Put(recordLink, result) return result, nil } func errorOrValue(err error, value interface{}) interface{} { if err == nil { return value } return err } // endpointToRecord puts the Target of an Endpoint in the correct field of DataBlock. // See DataBlock comments for more info func endpointToRecord(ep *endpoint.Endpoint) *dynect.DataBlock { result := dynect.DataBlock{} if ep.RecordType == endpoint.RecordTypeA { result.Address = ep.Targets[0] } else if ep.RecordType == endpoint.RecordTypeCNAME { result.CName = ep.Targets[0] } else if ep.RecordType == endpoint.RecordTypeTXT { result.TxtData = ep.Targets[0] } return &result } // fetchAllRecordLinksInZone list all records in a zone with a single call. Records not matched by the // DomainFilter are ignored. The response is a list of links that can be fed to dynect.Client.Do() // directly func (d *dynProviderState) fetchAllRecordLinksInZone(client *dynect.Client, zone string) ([]string, error) { var allRecords dynect.AllRecordsResponse err := client.Do("GET", fmt.Sprintf("AllRecord/%s/", zone), nil, &allRecords) if err != nil { return nil, err } return filterAndFixLinks(allRecords.Data, d.DomainFilter), nil } // buildLinkToRecord build a resource link. The symmetry of the dyn API is used to save // switch-case boilerplate. // Empty response means the endpoint is not mappable to a records link: either because the fqdn // is not matched by the domainFilter or it is in the wrong zone func (d *dynProviderState) buildLinkToRecord(ep *endpoint.Endpoint) string { if ep == nil { return "" } var matchingZone = "" for _, zone := range d.ZoneIDFilter.zoneIDs { if strings.HasSuffix(ep.DNSName, zone) { matchingZone = zone break } } if matchingZone == "" { // no matching zone, ignore return "" } if !d.DomainFilter.Match(ep.DNSName) { // no matching domain, ignore return "" } return fmt.Sprintf("%sRecord/%s/%s/", ep.RecordType, matchingZone, ep.DNSName) } // create a dynect client and performs login. You need to clean it up. // This method also stores the DynAPI version. // Don't user the dynect.Client.Login() func (d *dynProviderState) login() (*dynect.Client, error) { if d.LastLoginErrorTime != 0 { secondsSinceLastError := unixNow() - d.LastLoginErrorTime if secondsSinceLastError < badLoginMinIntervalSeconds { return nil, fmt.Errorf("will not attempt an API call as the last login failure occurred just %ds ago", secondsSinceLastError) } } client := dynect.NewClient(d.CustomerName) var req = dynect.LoginBlock{ Username: d.Username, Password: d.Password, CustomerName: d.CustomerName} var resp dynect.LoginResponse err := client.Do("POST", "Session", req, &resp) if err != nil { d.LastLoginErrorTime = unixNow() return nil, err } d.LastLoginErrorTime = 0 client.Token = resp.Data.Token // this is the only change from the original d.DynVersion = resp.Data.Version return client, nil } // the zones we are allowed to touch. Currently only exact matches are considered, not all // zones with the given suffix func (d *dynProviderState) zones(client *dynect.Client) []string { return d.ZoneIDFilter.zoneIDs } func (d *dynProviderState) buildRecordRequest(ep *endpoint.Endpoint) (string, *dynect.RecordRequest) { link := d.buildLinkToRecord(ep) if link == "" { return "", nil } record := dynect.RecordRequest{ TTL: fixMissingTTL(ep.RecordTTL, d.MinTTLSeconds), RData: *endpointToRecord(ep), } return link, &record } // deleteRecord deletes all existing records (CNAME, TXT, A) for the given Endpoint.DNSName with 1 API call func (d *dynProviderState) deleteRecord(client *dynect.Client, ep *endpoint.Endpoint) error { link := d.buildLinkToRecord(ep) if link == "" { return nil } response := dynect.RecordResponse{} err := client.Do("DELETE", link, nil, &response) log.Debugf("Deleting record %s: %+v,", link, errorOrValue(err, &response)) return err } // replaceRecord replaces all existing records pf the given type for the Endpoint.DNSName with 1 API call func (d *dynProviderState) replaceRecord(client *dynect.Client, ep *endpoint.Endpoint) error { link, record := d.buildRecordRequest(ep) if link == "" { return nil } response := dynect.RecordResponse{} err := client.Do("PUT", link, record, &response) log.Debugf("Replacing record %s: %+v,", link, errorOrValue(err, &response)) return err } // createRecord creates a single record with 1 API call func (d *dynProviderState) createRecord(client *dynect.Client, ep *endpoint.Endpoint) error { link, record := d.buildRecordRequest(ep) if link == "" { return nil } response := dynect.RecordResponse{} err := client.Do("POST", link, record, &response) log.Debugf("Creating record %s: %+v,", link, errorOrValue(err, &response)) return err } // commit commits all pending changes. It will always attempt to commit, if there are no func (d *dynProviderState) commit(client *dynect.Client) error { errs := []error{} for _, zone := range d.zones(client) { // extra call if in debug mode to fetch pending changes if log.GetLevel() >= log.DebugLevel { response := ZoneChangesResponse{} err := client.Do("GET", fmt.Sprintf("ZoneChanges/%s/", zone), nil, &response) log.Debugf("Pending changes for zone %s: %+v", zone, errorOrValue(err, &response)) } h, err := os.Hostname() if err != nil { h = "unknown-host" } notes := fmt.Sprintf("Change by external-dns@%s, DynAPI@%s, %s on %s", d.AppVersion, d.DynVersion, time.Now().Format(time.RFC3339), h, ) zonePublish := ZonePublishRequest{ Publish: true, Notes: notes, } response := ZonePublisResponse{} err = client.Do("PUT", fmt.Sprintf("Zone/%s/", zone), &zonePublish, &response) log.Infof("Commiting changes for zone %s: %+v", zone, errorOrValue(err, &response)) } switch len(errs) { case 0: return nil case 1: return errs[0] default: return fmt.Errorf("Multiple errors committing: %+v", errs) } } // Records makes on average C + 2*Z requests (Z = number of zones): 1 login + 1 fetchAllRecords // A cache is used to avoid querying for every single record found. C is proportional to the number // of expired/changed records func (d *dynProviderState) Records() ([]*endpoint.Endpoint, error) { client, err := d.login() if err != nil { return nil, err } defer client.Logout() log.Debugf("Using DynAPI@%s", d.DynVersion) var result []*endpoint.Endpoint zones := d.zones(client) log.Infof("Zones found: %+v", zones) for _, zone := range zones { recordLinks, err := d.fetchAllRecordLinksInZone(client, zone) if err != nil { return nil, err } log.Infof("Relevant records found in zone %s: %+v", zone, recordLinks) for _, link := range recordLinks { ep, err := d.recordLinkToEndpoint(client, link) if err != nil { return nil, err } if ep != nil { result = append(result, ep) } } } return result, nil } // this method does C + 2*Z requests: C=total number of changes, Z = number of // affected zones (1 login + 1 commit) func (d *dynProviderState) ApplyChanges(changes *plan.Changes) error { log.Debugf("Processing chages: %+v", changes) if d.DryRun { log.Infof("Will NOT delete these records: %+v", changes.Delete) log.Infof("Will NOT create these records: %+v", changes.Create) log.Infof("Will NOT update these records: %+v", merge(changes.UpdateOld, changes.UpdateNew)) return nil } client, err := d.login() if err != nil { return err } defer client.Logout() var errs []error needsCommit := false for _, ep := range changes.Delete { err := d.deleteRecord(client, ep) if err != nil { errs = append(errs, err) } else { needsCommit = true } } for _, ep := range changes.Create { err := d.createRecord(client, ep) if err != nil { errs = append(errs, err) } else { needsCommit = true } } updates := merge(changes.UpdateOld, changes.UpdateNew) log.Debugf("Updates after merging: %+v", updates) for _, ep := range updates { err := d.replaceRecord(client, ep) if err != nil { errs = append(errs, err) } else { needsCommit = true } } switch len(errs) { case 0: case 1: return errs[0] default: return fmt.Errorf("Multiple errors committing: %+v", errs) } if needsCommit { return d.commit(client) } return nil }