traefik/pkg/provider/kubernetes/gateway/client.go

package gateway

import (
	"context"
	"errors"
	"fmt"
	"os"
	"time"

	"github.com/rs/zerolog/log"
	corev1 "k8s.io/api/core/v1"
	kubeerror "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"sigs.k8s.io/gateway-api/apis/v1alpha2"
	"sigs.k8s.io/gateway-api/pkg/client/clientset/gateway/versioned"
	"sigs.k8s.io/gateway-api/pkg/client/informers/gateway/externalversions"
)

const resyncPeriod = 10 * time.Minute

type resourceEventHandler struct {
	ev chan<- interface{}
}

func (reh *resourceEventHandler) OnAdd(obj interface{}) {
	eventHandlerFunc(reh.ev, obj)
}

func (reh *resourceEventHandler) OnUpdate(oldObj, newObj interface{}) {
	switch oldObj.(type) {
	case *v1alpha2.GatewayClass:
		// Skip update for gateway classes. We only manage addition or deletion for this cluster-wide resource.
		return
	default:
		eventHandlerFunc(reh.ev, newObj)
	}
}

func (reh *resourceEventHandler) OnDelete(obj interface{}) {
	eventHandlerFunc(reh.ev, obj)
}

// Client is a client for the Provider master.
// WatchAll starts the watch of the Provider resources and updates the stores.
// The stores can then be accessed via the Get* functions.
type Client interface {
	WatchAll(namespaces []string, stopCh <-chan struct{}) (<-chan interface{}, error)
	GetGatewayClasses() ([]*v1alpha2.GatewayClass, error)
	UpdateGatewayStatus(gateway *v1alpha2.Gateway, gatewayStatus v1alpha2.GatewayStatus) error
	UpdateGatewayClassStatus(gatewayClass *v1alpha2.GatewayClass, condition metav1.Condition) error
	GetGateways() []*v1alpha2.Gateway
	GetHTTPRoutes(namespaces []string) ([]*v1alpha2.HTTPRoute, error)
	GetTCPRoutes(namespaces []string) ([]*v1alpha2.TCPRoute, error)
	GetTLSRoutes(namespaces []string) ([]*v1alpha2.TLSRoute, error)
	GetService(namespace, name string) (*corev1.Service, bool, error)
	GetSecret(namespace, name string) (*corev1.Secret, bool, error)
	GetEndpoints(namespace, name string) (*corev1.Endpoints, bool, error)
	GetNamespaces(selector labels.Selector) ([]string, error)
}

type clientWrapper struct {
	csGateway versioned.Interface
	csKube    kubernetes.Interface

	factoryNamespace    informers.SharedInformerFactory
	factoryGatewayClass externalversions.SharedInformerFactory
	factoriesGateway    map[string]externalversions.SharedInformerFactory
	factoriesKube       map[string]informers.SharedInformerFactory
	factoriesSecret     map[string]informers.SharedInformerFactory

	isNamespaceAll    bool
	watchedNamespaces []string

	labelSelector string
}

func createClientFromConfig(c *rest.Config) (*clientWrapper, error) {
	csGateway, err := versioned.NewForConfig(c)
	if err != nil {
		return nil, err
	}

	csKube, err := kubernetes.NewForConfig(c)
	if err != nil {
		return nil, err
	}

	return newClientImpl(csKube, csGateway), nil
}

func newClientImpl(csKube kubernetes.Interface, csGateway versioned.Interface) *clientWrapper {
	return &clientWrapper{
		csGateway:        csGateway,
		csKube:           csKube,
		factoriesGateway: make(map[string]externalversions.SharedInformerFactory),
		factoriesKube:    make(map[string]informers.SharedInformerFactory),
		factoriesSecret:  make(map[string]informers.SharedInformerFactory),
	}
}

// newInClusterClient returns a new Provider client that is expected to run
// inside the cluster.
func newInClusterClient(endpoint string) (*clientWrapper, error) {
	config, err := rest.InClusterConfig()
	if err != nil {
		return nil, fmt.Errorf("failed to create in-cluster configuration: %w", err)
	}

	if endpoint != "" {
		config.Host = endpoint
	}

	return createClientFromConfig(config)
}

func newExternalClusterClientFromFile(file string) (*clientWrapper, error) {
	configFromFlags, err := clientcmd.BuildConfigFromFlags("", file)
	if err != nil {
		return nil, err
	}
	return createClientFromConfig(configFromFlags)
}

// newExternalClusterClient returns a new Provider client that may run outside
// of the cluster.
// The endpoint parameter must not be empty.
func newExternalClusterClient(endpoint, token, caFilePath string) (*clientWrapper, error) {
	if endpoint == "" {
		return nil, errors.New("endpoint missing for external cluster client")
	}

	config := &rest.Config{
		Host:        endpoint,
		BearerToken: token,
	}

	if caFilePath != "" {
		caData, err := os.ReadFile(caFilePath)
		if err != nil {
			return nil, fmt.Errorf("failed to read CA file %s: %w", caFilePath, err)
		}

		config.TLSClientConfig = rest.TLSClientConfig{CAData: caData}
	}

	return createClientFromConfig(config)
}

// WatchAll starts namespace-specific controllers for all relevant kinds.
func (c *clientWrapper) WatchAll(namespaces []string, stopCh <-chan struct{}) (<-chan interface{}, error) {
	eventCh := make(chan interface{}, 1)
	eventHandler := &resourceEventHandler{ev: eventCh}

	if len(namespaces) == 0 {
		namespaces = []string{metav1.NamespaceAll}
		c.isNamespaceAll = true
	}

	c.watchedNamespaces = namespaces

	notOwnedByHelm := func(opts *metav1.ListOptions) {
		opts.LabelSelector = "owner!=helm"
	}

	labelSelectorOptions := func(options *metav1.ListOptions) {
		options.LabelSelector = c.labelSelector
	}

	c.factoryNamespace = informers.NewSharedInformerFactory(c.csKube, resyncPeriod)
	c.factoryNamespace.Core().V1().Namespaces().Informer().AddEventHandler(eventHandler)

	c.factoryGatewayClass = externalversions.NewSharedInformerFactoryWithOptions(c.csGateway, resyncPeriod, externalversions.WithTweakListOptions(labelSelectorOptions))
	c.factoryGatewayClass.Gateway().V1alpha2().GatewayClasses().Informer().AddEventHandler(eventHandler)

	// TODO manage Reference Policy
	// https://gateway-api.sigs.k8s.io/v1alpha2/references/spec/#gateway.networking.k8s.io/v1alpha2.ReferencePolicy

	for _, ns := range namespaces {
		factoryGateway := externalversions.NewSharedInformerFactoryWithOptions(c.csGateway, resyncPeriod, externalversions.WithNamespace(ns))
		factoryGateway.Gateway().V1alpha2().Gateways().Informer().AddEventHandler(eventHandler)
		factoryGateway.Gateway().V1alpha2().HTTPRoutes().Informer().AddEventHandler(eventHandler)
		factoryGateway.Gateway().V1alpha2().TCPRoutes().Informer().AddEventHandler(eventHandler)
		factoryGateway.Gateway().V1alpha2().TLSRoutes().Informer().AddEventHandler(eventHandler)

		factoryKube := informers.NewSharedInformerFactoryWithOptions(c.csKube, resyncPeriod, informers.WithNamespace(ns))
		factoryKube.Core().V1().Services().Informer().AddEventHandler(eventHandler)
		factoryKube.Core().V1().Endpoints().Informer().AddEventHandler(eventHandler)

		factorySecret := informers.NewSharedInformerFactoryWithOptions(c.csKube, resyncPeriod, informers.WithNamespace(ns), informers.WithTweakListOptions(notOwnedByHelm))
		factorySecret.Core().V1().Secrets().Informer().AddEventHandler(eventHandler)

		c.factoriesGateway[ns] = factoryGateway
		c.factoriesKube[ns] = factoryKube
		c.factoriesSecret[ns] = factorySecret
	}

	c.factoryNamespace.Start(stopCh)
	c.factoryGatewayClass.Start(stopCh)

	for _, ns := range namespaces {
		c.factoriesGateway[ns].Start(stopCh)
		c.factoriesKube[ns].Start(stopCh)
		c.factoriesSecret[ns].Start(stopCh)
	}

	for t, ok := range c.factoryNamespace.WaitForCacheSync(stopCh) {
		if !ok {
			return nil, fmt.Errorf("timed out waiting for controller caches to sync %s", t.String())
		}
	}

	for t, ok := range c.factoryGatewayClass.WaitForCacheSync(stopCh) {
		if !ok {
			return nil, fmt.Errorf("timed out waiting for controller caches to sync %s", t.String())
		}
	}

	for _, ns := range namespaces {
		for t, ok := range c.factoriesGateway[ns].WaitForCacheSync(stopCh) {
			if !ok {
				return nil, fmt.Errorf("timed out waiting for controller caches to sync %s in namespace %q", t.String(), ns)
			}
		}

		for t, ok := range c.factoriesKube[ns].WaitForCacheSync(stopCh) {
			if !ok {
				return nil, fmt.Errorf("timed out waiting for controller caches to sync %s in namespace %q", t.String(), ns)
			}
		}

		for t, ok := range c.factoriesSecret[ns].WaitForCacheSync(stopCh) {
			if !ok {
				return nil, fmt.Errorf("timed out waiting for controller caches to sync %s in namespace %q", t.String(), ns)
			}
		}
	}

	return eventCh, nil
}

func (c *clientWrapper) GetNamespaces(selector labels.Selector) ([]string, error) {
	ns, err := c.factoryNamespace.Core().V1().Namespaces().Lister().List(selector)
	if err != nil {
		return nil, err
	}

	var namespaces []string
	for _, namespace := range ns {
		if !c.isWatchedNamespace(namespace.Name) {
			log.Warn().Msgf("Namespace %q is not within %q watched namespaces", selector, namespace)
			continue
		}
		namespaces = append(namespaces, namespace.Name)
	}
	return namespaces, nil
}

func (c *clientWrapper) GetHTTPRoutes(namespaces []string) ([]*v1alpha2.HTTPRoute, error) {
	var httpRoutes []*v1alpha2.HTTPRoute
	for _, namespace := range namespaces {
		if !c.isWatchedNamespace(namespace) {
			log.Warn().Msgf("Failed to get HTTPRoutes: %q is not within watched namespaces", namespace)
			continue
		}

		routes, err := c.factoriesGateway[c.lookupNamespace(namespace)].Gateway().V1alpha2().HTTPRoutes().Lister().HTTPRoutes(namespace).List(labels.Everything())
		if err != nil {
			return nil, err
		}

		if len(routes) == 0 {
			log.Debug().Msgf("No HTTPRoutes found in namespace %q", namespace)
			continue
		}

		httpRoutes = append(httpRoutes, routes...)
	}

	return httpRoutes, nil
}

func (c *clientWrapper) GetTCPRoutes(namespaces []string) ([]*v1alpha2.TCPRoute, error) {
	var tcpRoutes []*v1alpha2.TCPRoute
	for _, namespace := range namespaces {
		if !c.isWatchedNamespace(namespace) {
			log.Warn().Msgf("Failed to get TCPRoutes: %q is not within watched namespaces", namespace)
			continue
		}

		routes, err := c.factoriesGateway[c.lookupNamespace(namespace)].Gateway().V1alpha2().TCPRoutes().Lister().TCPRoutes(namespace).List(labels.Everything())
		if err != nil {
			return nil, err
		}

		if len(routes) == 0 {
			log.Debug().Msgf("No TCPRoutes found in namespace %q", namespace)
			continue
		}

		tcpRoutes = append(tcpRoutes, routes...)
	}
	return tcpRoutes, nil
}

func (c *clientWrapper) GetTLSRoutes(namespaces []string) ([]*v1alpha2.TLSRoute, error) {
	var tlsRoutes []*v1alpha2.TLSRoute
	for _, namespace := range namespaces {
		if !c.isWatchedNamespace(namespace) {
			log.Warn().Msgf("Failed to get TLSRoutes: %q is not within watched namespaces", namespace)
			continue
		}

		routes, err := c.factoriesGateway[c.lookupNamespace(namespace)].Gateway().V1alpha2().TLSRoutes().Lister().TLSRoutes(namespace).List(labels.Everything())
		if err != nil {
			return nil, err
		}

		if len(routes) == 0 {
			log.Debug().Msgf("No TLSRoutes found in namespace %q", namespace)
			continue
		}

		tlsRoutes = append(tlsRoutes, routes...)
	}
	return tlsRoutes, nil
}

func (c *clientWrapper) GetGateways() []*v1alpha2.Gateway {
	var result []*v1alpha2.Gateway

	for ns, factory := range c.factoriesGateway {
		gateways, err := factory.Gateway().V1alpha2().Gateways().Lister().List(labels.Everything())
		if err != nil {
			log.Error().Err(err).Msgf("Failed to list Gateways in namespace %s", ns)
			continue
		}
		result = append(result, gateways...)
	}

	return result
}

func (c *clientWrapper) GetGatewayClasses() ([]*v1alpha2.GatewayClass, error) {
	return c.factoryGatewayClass.Gateway().V1alpha2().GatewayClasses().Lister().List(labels.Everything())
}

func (c *clientWrapper) UpdateGatewayClassStatus(gatewayClass *v1alpha2.GatewayClass, condition metav1.Condition) error {
	gc := gatewayClass.DeepCopy()

	var newConditions []metav1.Condition
	for _, cond := range gc.Status.Conditions {
		// No update for identical condition.
		if cond.Type == condition.Type && cond.Status == condition.Status {
			return nil
		}

		// Keep other condition types.
		if cond.Type != condition.Type {
			newConditions = append(newConditions, cond)
		}
	}

	// Append the condition to update.
	newConditions = append(newConditions, condition)
	gc.Status.Conditions = newConditions

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	_, err := c.csGateway.GatewayV1alpha2().GatewayClasses().UpdateStatus(ctx, gc, metav1.UpdateOptions{})
	if err != nil {
		return fmt.Errorf("failed to update GatewayClass %q status: %w", gatewayClass.Name, err)
	}

	return nil
}

func (c *clientWrapper) UpdateGatewayStatus(gateway *v1alpha2.Gateway, gatewayStatus v1alpha2.GatewayStatus) error {
	if !c.isWatchedNamespace(gateway.Namespace) {
		return fmt.Errorf("cannot update Gateway status %s/%s: namespace is not within watched namespaces", gateway.Namespace, gateway.Name)
	}

	if statusEquals(gateway.Status, gatewayStatus) {
		return nil
	}

	g := gateway.DeepCopy()
	g.Status = gatewayStatus

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	_, err := c.csGateway.GatewayV1alpha2().Gateways(gateway.Namespace).UpdateStatus(ctx, g, metav1.UpdateOptions{})
	if err != nil {
		return fmt.Errorf("failed to update Gateway %q status: %w", gateway.Name, err)
	}

	return nil
}

func statusEquals(oldStatus, newStatus v1alpha2.GatewayStatus) bool {
	if len(oldStatus.Listeners) != len(newStatus.Listeners) {
		return false
	}

	if !conditionsEquals(oldStatus.Conditions, newStatus.Conditions) {
		return false
	}

	listenerMatches := 0
	for _, newListener := range newStatus.Listeners {
		for _, oldListener := range oldStatus.Listeners {
			if newListener.Name == oldListener.Name {
				if !conditionsEquals(newListener.Conditions, oldListener.Conditions) {
					return false
				}

				listenerMatches++
			}
		}
	}

	return listenerMatches == len(oldStatus.Listeners)
}

func conditionsEquals(conditionsA, conditionsB []metav1.Condition) bool {
	if len(conditionsA) != len(conditionsB) {
		return false
	}

	conditionMatches := 0
	for _, conditionA := range conditionsA {
		for _, conditionB := range conditionsB {
			if conditionA.Type == conditionB.Type {
				if conditionA.Reason != conditionB.Reason || conditionA.Status != conditionB.Status || conditionA.Message != conditionB.Message {
					return false
				}
				conditionMatches++
			}
		}
	}

	return conditionMatches == len(conditionsA)
}

// GetService returns the named service from the given namespace.
func (c *clientWrapper) GetService(namespace, name string) (*corev1.Service, bool, error) {
	if !c.isWatchedNamespace(namespace) {
		return nil, false, fmt.Errorf("failed to get service %s/%s: namespace is not within watched namespaces", namespace, name)
	}

	service, err := c.factoriesKube[c.lookupNamespace(namespace)].Core().V1().Services().Lister().Services(namespace).Get(name)
	exist, err := translateNotFoundError(err)

	return service, exist, err
}

// GetEndpoints returns the named endpoints from the given namespace.
func (c *clientWrapper) GetEndpoints(namespace, name string) (*corev1.Endpoints, bool, error) {
	if !c.isWatchedNamespace(namespace) {
		return nil, false, fmt.Errorf("failed to get endpoints %s/%s: namespace is not within watched namespaces", namespace, name)
	}

	endpoint, err := c.factoriesKube[c.lookupNamespace(namespace)].Core().V1().Endpoints().Lister().Endpoints(namespace).Get(name)
	exist, err := translateNotFoundError(err)

	return endpoint, exist, err
}

// GetSecret returns the named secret from the given namespace.
func (c *clientWrapper) GetSecret(namespace, name string) (*corev1.Secret, bool, error) {
	if !c.isWatchedNamespace(namespace) {
		return nil, false, fmt.Errorf("failed to get secret %s/%s: namespace is not within watched namespaces", namespace, name)
	}

	secret, err := c.factoriesSecret[c.lookupNamespace(namespace)].Core().V1().Secrets().Lister().Secrets(namespace).Get(name)
	exist, err := translateNotFoundError(err)

	return secret, exist, err
}

// lookupNamespace returns the lookup namespace key for the given namespace.
// When listening on all namespaces, it returns the client-go identifier ("")
// for all-namespaces. Otherwise, it returns the given namespace.
// The distinction is necessary because we index all informers on the special
// identifier iff all-namespaces are requested but receive specific namespace
// identifiers from the Kubernetes API, so we have to bridge this gap.
func (c *clientWrapper) lookupNamespace(ns string) string {
	if c.isNamespaceAll {
		return metav1.NamespaceAll
	}
	return ns
}

// eventHandlerFunc will pass the obj on to the events channel or drop it.
// This is so passing the events along won't block in the case of high volume.
// The events are only used for signaling anyway so dropping a few is ok.
func eventHandlerFunc(events chan<- interface{}, obj interface{}) {
	select {
	case events <- obj:
	default:
	}
}

// translateNotFoundError will translate a "not found" error to a boolean return
// value which indicates if the resource exists and a nil error.
func translateNotFoundError(err error) (bool, error) {
	if kubeerror.IsNotFound(err) {
		return false, nil
	}
	return err == nil, err
}

// isWatchedNamespace checks to ensure that the namespace is being watched before we request
// it to ensure we don't panic by requesting an out-of-watch object.
func (c *clientWrapper) isWatchedNamespace(ns string) bool {
	if c.isNamespaceAll {
		return true
	}
	for _, watchedNamespace := range c.watchedNamespaces {
		if watchedNamespace == ns {
			return true
		}
	}
	return false
}