external-dns/source/f5_transportserver.go

/*
Copyright 2022 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"
	"errors"
	"fmt"
	"strings"

	log "github.com/sirupsen/logrus"
	"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/client-go/dynamic"
	"k8s.io/client-go/dynamic/dynamicinformer"
	kubeinformers "k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/tools/cache"

	f5 "github.com/F5Networks/k8s-bigip-ctlr/v2/config/apis/cis/v1"

	"sigs.k8s.io/external-dns/source/informers"

	"sigs.k8s.io/external-dns/endpoint"
	"sigs.k8s.io/external-dns/source/annotations"
)

var f5TransportServerGVR = schema.GroupVersionResource{
	Group:    "cis.f5.com",
	Version:  "v1",
	Resource: "transportservers",
}

// transportServerSource is an implementation of Source for F5 TransportServer objects.
type f5TransportServerSource struct {
	dynamicKubeClient       dynamic.Interface
	transportServerInformer kubeinformers.GenericInformer
	kubeClient              kubernetes.Interface
	annotationFilter        string
	namespace               string
	unstructuredConverter   *unstructuredConverter
}

func NewF5TransportServerSource(
	ctx context.Context,
	dynamicKubeClient dynamic.Interface,
	kubeClient kubernetes.Interface,
	namespace string,
	annotationFilter string,
) (Source, error) {
	informerFactory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicKubeClient, 0, namespace, nil)
	transportServerInformer := informerFactory.ForResource(f5TransportServerGVR)

	transportServerInformer.Informer().AddEventHandler(
		cache.ResourceEventHandlerFuncs{
			AddFunc: func(obj interface{}) {
			},
		},
	)

	informerFactory.Start(ctx.Done())

	// wait for the local cache to be populated.
	if err := informers.WaitForDynamicCacheSync(context.Background(), informerFactory); err != nil {
		return nil, err
	}

	uc, err := newTSUnstructuredConverter()
	if err != nil {
		return nil, fmt.Errorf("failed to setup unstructured converter: %w", err)
	}

	return &f5TransportServerSource{
		dynamicKubeClient:       dynamicKubeClient,
		transportServerInformer: transportServerInformer,
		kubeClient:              kubeClient,
		namespace:               namespace,
		annotationFilter:        annotationFilter,
		unstructuredConverter:   uc,
	}, nil
}

// Endpoints returns endpoint objects for each host-target combination that should be processed.
// Retrieves all TransportServers in the source's namespace(s).
func (ts *f5TransportServerSource) Endpoints(ctx context.Context) ([]*endpoint.Endpoint, error) {
	transportServerObjects, err := ts.transportServerInformer.Lister().ByNamespace(ts.namespace).List(labels.Everything())
	if err != nil {
		return nil, err
	}

	var transportServers []*f5.TransportServer
	for _, tsObj := range transportServerObjects {
		unstructuredHost, ok := tsObj.(*unstructured.Unstructured)
		if !ok {
			return nil, errors.New("could not convert")
		}

		transportServer := &f5.TransportServer{}
		err := ts.unstructuredConverter.scheme.Convert(unstructuredHost, transportServer, nil)
		if err != nil {
			return nil, err
		}
		transportServers = append(transportServers, transportServer)
	}

	transportServers, err = ts.filterByAnnotations(transportServers)
	if err != nil {
		return nil, fmt.Errorf("failed to filter TransportServers: %w", err)
	}

	endpoints, err := ts.endpointsFromTransportServers(transportServers)
	if err != nil {
		return nil, err
	}

	return endpoints, nil
}

func (ts *f5TransportServerSource) AddEventHandler(ctx context.Context, handler func()) {
	log.Debug("Adding event handler for TransportServer")

	ts.transportServerInformer.Informer().AddEventHandler(eventHandlerFunc(handler))
}

// endpointsFromTransportServers extracts the endpoints from a slice of TransportServers
func (ts *f5TransportServerSource) endpointsFromTransportServers(transportServers []*f5.TransportServer) ([]*endpoint.Endpoint, error) {
	var endpoints []*endpoint.Endpoint

	for _, transportServer := range transportServers {
		if !hasValidTransportServerIP(transportServer) {
			log.Warnf("F5 TransportServer %s/%s is missing a valid IP address, skipping endpoint creation.",
				transportServer.Namespace, transportServer.Name)
			continue
		}

		resource := fmt.Sprintf("f5-transportserver/%s/%s", transportServer.Namespace, transportServer.Name)

		ttl := annotations.TTLFromAnnotations(transportServer.Annotations, resource)

		targets := annotations.TargetsFromTargetAnnotation(transportServer.Annotations)
		if len(targets) == 0 && transportServer.Spec.VirtualServerAddress != "" {
			targets = append(targets, transportServer.Spec.VirtualServerAddress)
		}
		if len(targets) == 0 && transportServer.Status.VSAddress != "" {
			targets = append(targets, transportServer.Status.VSAddress)
		}

		endpoints = append(endpoints, EndpointsForHostname(transportServer.Spec.Host, targets, ttl, nil, "", resource)...)
	}

	return endpoints, nil
}

// newUnstructuredConverter returns a new unstructuredConverter initialized
func newTSUnstructuredConverter() (*unstructuredConverter, error) {
	uc := &unstructuredConverter{
		scheme: runtime.NewScheme(),
	}

	// Add the core types we need
	uc.scheme.AddKnownTypes(f5TransportServerGVR.GroupVersion(), &f5.TransportServer{}, &f5.TransportServerList{})
	if err := scheme.AddToScheme(uc.scheme); err != nil {
		return nil, err
	}

	return uc, nil
}

// filterByAnnotations filters a list of TransportServers by a given annotation selector.
func (ts *f5TransportServerSource) filterByAnnotations(transportServers []*f5.TransportServer) ([]*f5.TransportServer, error) {
	selector, err := annotations.ParseFilter(ts.annotationFilter)
	if err != nil {
		return nil, err
	}

	// empty filter returns original list
	if selector.Empty() {
		return transportServers, nil
	}

	filteredList := []*f5.TransportServer{}

	for _, ts := range transportServers {
		// include TransportServer if its annotations match the selector
		if selector.Matches(labels.Set(ts.Annotations)) {
			filteredList = append(filteredList, ts)
		}
	}

	return filteredList, nil
}

func hasValidTransportServerIP(vs *f5.TransportServer) bool {
	normalizedAddress := strings.ToLower(vs.Status.VSAddress)
	return normalizedAddress != "none" && normalizedAddress != ""
}