Merge pull request #1642 from datawire/inercia/ambassador_host

Support Ambassador Host resources as sources

go.mod

@@ -8,7 +8,6 @@ require (
 	github.com/Azure/azure-sdk-for-go v45.1.0+incompatible
 	github.com/Azure/go-autorest/autorest v0.11.10
 	github.com/Azure/go-autorest/autorest/adal v0.9.5
-	github.com/Azure/go-autorest/autorest/azure/auth v0.5.3
 	github.com/Azure/go-autorest/autorest/to v0.4.0
 	github.com/akamai/AkamaiOPEN-edgegrid-golang v1.0.0
 	github.com/alecthomas/assert v0.0.0-20170929043011-405dbfeb8e38 // indirect
@@ -19,6 +18,7 @@ require (
 	github.com/aws/aws-sdk-go v1.31.4
 	github.com/cloudflare/cloudflare-go v0.10.1
 	github.com/cloudfoundry-community/go-cfclient v0.0.0-20190201205600-f136f9222381
+	github.com/datawire/ambassador v1.6.0
 	github.com/denverdino/aliyungo v0.0.0-20190125010748-a747050bb1ba
 	github.com/digitalocean/godo v1.36.0
 	github.com/dnsimple/dnsimple-go v0.60.0
@@ -46,7 +46,8 @@ require (
 	github.com/sanyu/dynectsoap v0.0.0-20181203081243-b83de5edc4e0
 	github.com/scaleway/scaleway-sdk-go v1.0.0-beta.6.0.20200623155123-84df6c4b5301
 	github.com/sirupsen/logrus v1.6.0
-	github.com/stretchr/testify v1.5.1
+	github.com/smartystreets/gunit v1.3.4 // indirect
+	github.com/stretchr/testify v1.6.1
 	github.com/terra-farm/udnssdk v1.3.5 // indirect
 	github.com/transip/gotransip v5.8.2+incompatible
 	github.com/ultradns/ultradns-sdk-go v0.0.0-20200616202852-e62052662f60
@@ -54,16 +55,19 @@ require (
 	github.com/vultr/govultr v0.4.2
 	go.etcd.io/etcd v0.5.0-alpha.5.0.20200401174654-e694b7bb0875
 	go.uber.org/ratelimit v0.1.0
-	golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e
+	golang.org/x/net v0.0.0-20200625001655-4c5254603344
 	golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d
+	golang.org/x/tools v0.0.0-20200708003708-134513de8882 // indirect
 	google.golang.org/api v0.15.0
 	gopkg.in/ns1/ns1-go.v2 v2.0.0-20190322154155-0dafb5275fd1
-	gopkg.in/yaml.v2 v2.2.8
+	gopkg.in/yaml.v2 v2.3.0
+	honnef.co/go/tools v0.0.1-2020.1.4 // indirect
 	istio.io/api v0.0.0-20200529165953-72dad51d4ffc
 	istio.io/client-go v0.0.0-20200529172309-31c16ea3f751
 	k8s.io/api v0.18.8
 	k8s.io/apimachinery v0.18.8
 	k8s.io/client-go v0.18.8
+	k8s.io/kubernetes v1.13.0
 )
 
 replace (

pkg/apis/externaldns/types.go

@@ -318,7 +318,7 @@ func (cfg *Config) ParseFlags(args []string) error {
 	app.Flag("skipper-routegroup-groupversion", "The resource version for skipper routegroup").Default(source.DefaultRoutegroupVersion).StringVar(&cfg.SkipperRouteGroupVersion)
 
 	// Flags related to processing sources
-	app.Flag("source", "The resource types that are queried for endpoints; specify multiple times for multiple sources (required, options: service, ingress, node, fake, connector, istio-gateway, istio-virtualservice, cloudfoundry, contour-ingressroute, contour-httpproxy, crd, empty, skipper-routegroup,openshift-route)").Required().PlaceHolder("source").EnumsVar(&cfg.Sources, "service", "ingress", "node", "istio-gateway", "istio-virtualservice", "cloudfoundry", "contour-ingressroute", "contour-httpproxy", "fake", "connector", "crd", "empty", "skipper-routegroup", "openshift-route")
+	app.Flag("source", "The resource types that are queried for endpoints; specify multiple times for multiple sources (required, options: service, ingress, node, fake, connector, istio-gateway, istio-virtualservice, cloudfoundry, contour-ingressroute, contour-httpproxy, crd, empty, skipper-routegroup, openshift-route, ambassador-host)").Required().PlaceHolder("source").EnumsVar(&cfg.Sources, "service", "ingress", "node", "istio-gateway", "istio-virtualservice", "cloudfoundry", "contour-ingressroute", "contour-httpproxy", "fake", "connector", "crd", "empty", "skipper-routegroup", "openshift-route", "ambassador-host")
 
 	app.Flag("namespace", "Limit sources of endpoints to a specific namespace (default: all namespaces)").Default(defaultConfig.Namespace).StringVar(&cfg.Namespace)
 	app.Flag("annotation-filter", "Filter sources managed by external-dns via annotation using label selector semantics (default: all sources)").Default(defaultConfig.AnnotationFilter).StringVar(&cfg.AnnotationFilter)

source/ambassador_host.go

@@ -0,0 +1,283 @@
+/*
+Copyright 2020 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 source
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	ambassador "github.com/datawire/ambassador/pkg/api/getambassador.io/v2"
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/dynamic/dynamicinformer"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/cache"
+	api "k8s.io/kubernetes/pkg/apis/core"
+
+	"sigs.k8s.io/external-dns/endpoint"
+)
+
+// ambHostAnnotation is the annotation in the Host that maps to a Service
+const ambHostAnnotation = "external-dns.ambassador-service"
+
+// groupName is the group name for the Ambassador API
+const groupName = "getambassador.io"
+
+var schemeGroupVersion = schema.GroupVersion{Group: groupName, Version: "v2"}
+
+var ambHostGVR = schemeGroupVersion.WithResource("hosts")
+
+// ambassadorHostSource is an implementation of Source for Ambassador Host objects.
+// The IngressRoute implementation uses the spec.virtualHost.fqdn value for the hostname.
+// Use targetAnnotationKey to explicitly set Endpoint.
+type ambassadorHostSource struct {
+	dynamicKubeClient      dynamic.Interface
+	kubeClient             kubernetes.Interface
+	namespace              string
+	ambassadorHostInformer informers.GenericInformer
+	unstructuredConverter  *unstructuredConverter
+}
+
+// NewAmbassadorHostSource creates a new ambassadorHostSource with the given config.
+func NewAmbassadorHostSource(
+	dynamicKubeClient dynamic.Interface,
+	kubeClient kubernetes.Interface,
+	namespace string) (Source, error) {
+	var err error
+
+	// Use shared informer to listen for add/update/delete of Host in the specified namespace.
+	// Set resync period to 0, to prevent processing when nothing has changed.
+	informerFactory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicKubeClient, 0, namespace, nil)
+	ambassadorHostInformer := informerFactory.ForResource(ambHostGVR)
+
+	// Add default resource event handlers to properly initialize informer.
+	ambassadorHostInformer.Informer().AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+			},
+		},
+	)
+
+	// TODO informer is not explicitly stopped since controller is not passing in its channel.
+	informerFactory.Start(wait.NeverStop)
+
+	// wait for the local cache to be populated.
+	err = poll(time.Second, 60*time.Second, func() (bool, error) {
+		return ambassadorHostInformer.Informer().HasSynced(), nil
+	})
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to sync cache")
+	}
+
+	uc, err := newUnstructuredConverter()
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to setup Unstructured Converter")
+	}
+
+	return &ambassadorHostSource{
+		dynamicKubeClient:      dynamicKubeClient,
+		kubeClient:             kubeClient,
+		namespace:              namespace,
+		ambassadorHostInformer: ambassadorHostInformer,
+		unstructuredConverter:  uc,
+	}, nil
+}
+
+// Endpoints returns endpoint objects for each host-target combination that should be processed.
+// Retrieves all Hosts in the source's namespace(s).
+func (sc *ambassadorHostSource) Endpoints(ctx context.Context) ([]*endpoint.Endpoint, error) {
+	hosts, err := sc.ambassadorHostInformer.Lister().ByNamespace(sc.namespace).List(labels.Everything())
+	if err != nil {
+		return nil, err
+	}
+
+	endpoints := []*endpoint.Endpoint{}
+	for _, hostObj := range hosts {
+		unstructuredHost, ok := hostObj.(*unstructured.Unstructured)
+		if !ok {
+			return nil, errors.New("could not convert")
+		}
+
+		host := &ambassador.Host{}
+		err := sc.unstructuredConverter.scheme.Convert(unstructuredHost, host, nil)
+		if err != nil {
+			return nil, err
+		}
+
+		fullname := fmt.Sprintf("%s/%s", host.Namespace, host.Name)
+
+		// look for the "exernal-dns.ambassador-service" annotation. If it is not there then just ignore this `Host`
+		service, found := host.Annotations[ambHostAnnotation]
+		if !found {
+			log.Debugf("Host %s ignored: no annotation %q found", fullname, ambHostAnnotation)
+			continue
+		}
+
+		targets, err := sc.targetsFromAmbassadorLoadBalancer(ctx, service)
+		if err != nil {
+			return nil, err
+		}
+
+		hostEndpoints, err := sc.endpointsFromHost(ctx, host, targets)
+		if err != nil {
+			return nil, err
+		}
+		if len(hostEndpoints) == 0 {
+			log.Debugf("No endpoints could be generated from Host %s", fullname)
+			continue
+		}
+
+		log.Debugf("Endpoints generated from Host: %s: %v", fullname, hostEndpoints)
+		endpoints = append(endpoints, hostEndpoints...)
+	}
+
+	for _, ep := range endpoints {
+		sort.Sort(ep.Targets)
+	}
+
+	return endpoints, nil
+}
+
+// endpointsFromHost extracts the endpoints from a Host object
+func (sc *ambassadorHostSource) endpointsFromHost(ctx context.Context, host *ambassador.Host, targets endpoint.Targets) ([]*endpoint.Endpoint, error) {
+	var endpoints []*endpoint.Endpoint
+
+	providerSpecific := endpoint.ProviderSpecific{}
+	setIdentifier := ""
+
+	annotations := host.Annotations
+	ttl, err := getTTLFromAnnotations(annotations)
+	if err != nil {
+		return nil, err
+	}
+
+	if host.Spec != nil {
+		hostname := host.Spec.Hostname
+		if hostname != "" {
+			endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific, setIdentifier)...)
+		}
+	}
+
+	return endpoints, nil
+}
+
+func (sc *ambassadorHostSource) targetsFromAmbassadorLoadBalancer(ctx context.Context, service string) (targets endpoint.Targets, err error) {
+	lbNamespace, lbName, err := parseAmbLoadBalancerService(service)
+	if err != nil {
+		return nil, err
+	}
+
+	svc, err := sc.kubeClient.CoreV1().Services(lbNamespace).Get(ctx, lbName, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	for _, lb := range svc.Status.LoadBalancer.Ingress {
+		if lb.IP != "" {
+			targets = append(targets, lb.IP)
+		}
+		if lb.Hostname != "" {
+			targets = append(targets, lb.Hostname)
+		}
+	}
+
+	return
+}
+
+// parseAmbLoadBalancerService returns a name/namespace tuple from the annotation in
+// an Ambassador Host CRD
+//
+// This is a thing because Ambassador has historically supported cross-namespace
+// references using a name.namespace syntax, but here we want to also support
+// namespace/name.
+//
+// Returns namespace, name, error.
+
+func parseAmbLoadBalancerService(service string) (namespace, name string, err error) {
+	// Start by assuming that we have namespace/name.
+	parts := strings.Split(service, "/")
+
+	if len(parts) == 1 {
+		// No "/" at all, so let's try for name.namespace. To be consistent with the
+		// rest of Ambassador, use SplitN to limit this to one split, so that e.g.
+		// svc.foo.bar uses service "svc" in namespace "foo.bar".
+		parts = strings.SplitN(service, ".", 2)
+
+		if len(parts) == 2 {
+			// We got a namespace, great.
+			name := parts[0]
+			namespace := parts[1]
+
+			return namespace, name, nil
+		}
+
+		// If here, we have no separator, so the whole string is the service, and
+		// we can assume the default namespace.
+		name := service
+		namespace := api.NamespaceDefault
+
+		return namespace, name, nil
+	} else if len(parts) == 2 {
+		// This is "namespace/name". Note that the name could be qualified,
+		// which is fine.
+		namespace := parts[0]
+		name := parts[1]
+
+		return namespace, name, nil
+	}
+
+	// If we got here, this string is simply ill-formatted. Return an error.
+	return "", "", errors.New(fmt.Sprintf("invalid external-dns service: %s", service))
+}
+
+func (sc *ambassadorHostSource) AddEventHandler(ctx context.Context, handler func()) {
+}
+
+// unstructuredConverter handles conversions between unstructured.Unstructured and Ambassador types
+type unstructuredConverter struct {
+	// scheme holds an initializer for converting Unstructured to a type
+	scheme *runtime.Scheme
+}
+
+// newUnstructuredConverter returns a new unstructuredConverter initialized
+func newUnstructuredConverter() (*unstructuredConverter, error) {
+	uc := &unstructuredConverter{
+		scheme: runtime.NewScheme(),
+	}
+
+	// Setup converter to understand custom CRD types
+	ambassador.AddToScheme(uc.scheme)
+
+	// Add the core types we need
+	if err := scheme.AddToScheme(uc.scheme); err != nil {
+		return nil, err
+	}
+
+	return uc, nil
+}

source/ambassador_host_test.go

@@ -0,0 +1,78 @@
+/*
+Copyright 2019 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 source
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+)
+
+type AmbassadorSuite struct {
+	suite.Suite
+}
+
+func TestAmbassadorSource(t *testing.T) {
+	suite.Run(t, new(AmbassadorSuite))
+	t.Run("Interface", testAmbassadorSourceImplementsSource)
+}
+
+// testAmbassadorSourceImplementsSource tests that ambassadorHostSource is a valid Source.
+func testAmbassadorSourceImplementsSource(t *testing.T) {
+	require.Implements(t, (*Source)(nil), new(ambassadorHostSource))
+}
+
+// TestParseAmbLoadBalancerService tests our parsing of Ambassador service info.
+func TestParseAmbLoadBalancerService(t *testing.T) {
+	vectors := []struct {
+		input  string
+		ns     string
+		svc    string
+		errstr string
+	}{
+		{"svc", "default", "svc", ""},
+		{"ns/svc", "ns", "svc", ""},
+		{"svc.ns", "ns", "svc", ""},
+		{"svc.ns.foo.bar", "ns.foo.bar", "svc", ""},
+		{"ns/svc/foo/bar", "", "", "invalid external-dns service: ns/svc/foo/bar"},
+		{"ns/svc/foo.bar", "", "", "invalid external-dns service: ns/svc/foo.bar"},
+		{"ns.foo/svc/bar", "", "", "invalid external-dns service: ns.foo/svc/bar"},
+	}
+
+	for _, v := range vectors {
+		ns, svc, err := parseAmbLoadBalancerService(v.input)
+
+		errstr := ""
+
+		if err != nil {
+			errstr = err.Error()
+		}
+
+		if v.ns != ns {
+			t.Errorf("%s: got ns \"%s\", wanted \"%s\"", v.input, ns, v.ns)
+		}
+
+		if v.svc != svc {
+			t.Errorf("%s: got svc \"%s\", wanted \"%s\"", v.input, svc, v.svc)
+		}
+
+		if v.errstr != errstr {
+			t.Errorf("%s: got err \"%s\", wanted \"%s\"", v.input, errstr, v.errstr)
+		}
+	}
+}

source/store.go

@@ -83,12 +83,12 @@ type SingletonClientGenerator struct {
 	kubeClient      kubernetes.Interface
 	istioClient     *istioclient.Clientset
 	cfClient        *cfclient.Client
-	contourClient   dynamic.Interface
+	dynKubeClient   dynamic.Interface
 	openshiftClient openshift.Interface
 	kubeOnce        sync.Once
 	istioOnce       sync.Once
 	cfOnce          sync.Once
-	contourOnce     sync.Once
+	dynCliOnce      sync.Once
 	openshiftOnce   sync.Once
 }
 
@@ -134,13 +134,13 @@ func NewCFClient(cfAPIEndpoint string, cfUsername string, cfPassword string) (*c
 	return client, nil
 }
 
-// DynamicKubernetesClient generates a contour client if it was not created before
+// DynamicKubernetesClient generates a dynamic client if it was not created before
 func (p *SingletonClientGenerator) DynamicKubernetesClient() (dynamic.Interface, error) {
 	var err error
-	p.contourOnce.Do(func() {
+	p.dynCliOnce.Do(func() {
-		p.contourClient, err = NewDynamicKubernetesClient(p.KubeConfig, p.APIServerURL, p.RequestTimeout)
+		p.dynKubeClient, err = NewDynamicKubernetesClient(p.KubeConfig, p.APIServerURL, p.RequestTimeout)
 	})
-	return p.contourClient, err
+	return p.dynKubeClient, err
 }
 
 // OpenShiftClient generates an openshift client if it was not created before
@@ -213,6 +213,16 @@ func BuildWithConfig(source string, p ClientGenerator, cfg *Config) (Source, err
 			return nil, err
 		}
 		return NewCloudFoundrySource(cfClient)
+	case "ambassador-host":
+		kubernetesClient, err := p.KubeClient()
+		if err != nil {
+			return nil, err
+		}
+		dynamicClient, err := p.DynamicKubernetesClient()
+		if err != nil {
+			return nil, err
+		}
+		return NewAmbassadorHostSource(dynamicClient, kubernetesClient, cfg.Namespace)
 	case "contour-ingressroute":
 		kubernetesClient, err := p.KubeClient()
 		if err != nil {