ingress/pkg/provider/aggregator/aggregator.go

package aggregator

import (
	"context"
	"fmt"
	"time"

	"github.com/rs/zerolog/log"
	"github.com/hanzoai/ingress/pkg/config/dynamic"
	"github.com/hanzoai/ingress/pkg/config/static"
	"github.com/hanzoai/ingress/pkg/provider"
	"github.com/hanzoai/ingress/pkg/provider/file"
	"github.com/hanzoai/ingress/pkg/provider/builtins"
	"github.com/hanzoai/ingress/pkg/redactor"
	"github.com/hanzoai/ingress/pkg/safe"
)

// throttled defines what kind of config refresh throttling the aggregator should
// set up for a given provider.
// If a provider implements throttled, the configuration changes it sends will be
// taken into account no more often than the frequency inferred from ThrottleDuration().
// If ThrottleDuration returns zero, no throttling will take place.
// If throttled is not implemented, the throttling will be set up in accordance
// with the global providersThrottleDuration option.
type throttled interface {
	ThrottleDuration() time.Duration
}

// maybeThrottledProvide returns the Provide method of the given provider,
// potentially augmented with some throttling depending on whether and how the
// provider implements the throttled interface.
func maybeThrottledProvide(prd provider.Provider, defaultDuration time.Duration) func(chan<- dynamic.Message, *safe.Pool) error {
	providerThrottleDuration := defaultDuration
	if throttled, ok := prd.(throttled); ok {
		// per-provider throttling
		providerThrottleDuration = throttled.ThrottleDuration()
	}

	if providerThrottleDuration == 0 {
		// throttling disabled
		return prd.Provide
	}

	return func(configurationChan chan<- dynamic.Message, pool *safe.Pool) error {
		rc := newRingChannel()
		pool.GoCtx(func(ctx context.Context) {
			for {
				select {
				case <-ctx.Done():
					return
				case msg := <-rc.out():
					configurationChan <- msg
					time.Sleep(providerThrottleDuration)
				}
			}
		})

		return prd.Provide(rc.in(), pool)
	}
}

// ProviderAggregator aggregates providers.
type ProviderAggregator struct {
	internalProvider          provider.Provider
	fileProvider              provider.Provider
	providers                 []provider.Provider
	providersThrottleDuration time.Duration
}

// NewProviderAggregator returns an aggregate of all the providers configured in the static configuration.
func NewProviderAggregator(conf static.Providers) *ProviderAggregator {
	p := &ProviderAggregator{
		providersThrottleDuration: time.Duration(conf.ProvidersThrottleDuration),
	}

	if conf.File != nil {
		p.quietAddProvider(conf.File)
	}

	if conf.KubernetesIngress != nil {
		p.quietAddProvider(conf.KubernetesIngress)
	}

	if conf.KubernetesIngressNGINX != nil {
		p.quietAddProvider(conf.KubernetesIngressNGINX)
	}

	if conf.KubernetesCRD != nil {
		p.quietAddProvider(conf.KubernetesCRD)
	}

	if conf.Knative != nil {
		p.quietAddProvider(conf.Knative)
	}

	if conf.KubernetesGateway != nil {
		p.quietAddProvider(conf.KubernetesGateway)
	}

	return p
}

// AddProvider adds a provider in the providers map.
func (p *ProviderAggregator) AddProvider(provider provider.Provider) error {
	err := provider.Init()
	if err != nil {
		return err
	}

	switch provider.(type) {
	case *file.Provider:
		p.fileProvider = provider
	case *builtins.Provider:
		p.internalProvider = provider
	default:
		p.providers = append(p.providers, provider)
	}

	return nil
}

// Init the provider.
func (p *ProviderAggregator) Init() error {
	return nil
}

// Provide calls the provide method of every providers.
func (p *ProviderAggregator) Provide(configurationChan chan<- dynamic.Message, pool *safe.Pool) error {
	if p.fileProvider != nil {
		p.launchProvider(configurationChan, pool, p.fileProvider)
	}

	for _, prd := range p.providers {
		safe.Go(func() {
			p.launchProvider(configurationChan, pool, prd)
		})
	}

	// internal provider must be the last because we use it to know if all the providers are loaded.
	// ConfigurationWatcher will wait for this requiredProvider before applying configurations.
	if p.internalProvider != nil {
		p.launchProvider(configurationChan, pool, p.internalProvider)
	}

	return nil
}

func (p *ProviderAggregator) quietAddProvider(provider provider.Provider) {
	err := p.AddProvider(provider)
	if err != nil {
		log.Error().Err(err).Msgf("Error while initializing provider %T", provider)
	}
}

func (p *ProviderAggregator) launchProvider(configurationChan chan<- dynamic.Message, pool *safe.Pool, prd provider.Provider) {
	jsonConf, err := redactor.RemoveCredentials(prd)
	if err != nil {
		log.Debug().Err(err).Msgf("Cannot marshal the provider configuration %T", prd)
	}

	// Check if provider has namespace information.
	var namespaceInfo string
	if namespaceProvider, ok := prd.(provider.NamespacedProvider); ok {
		if namespace := namespaceProvider.Namespace(); namespace != "" {
			namespaceInfo = fmt.Sprintf(" (namespace: %s)", namespace)
		}
	}

	log.Info().Msgf("Starting provider %T%s", prd, namespaceInfo)
	log.Debug().RawJSON("config", []byte(jsonConf)).Msgf("%T provider configuration%s", prd, namespaceInfo)

	if err := maybeThrottledProvide(prd, p.providersThrottleDuration)(configurationChan, pool); err != nil {
		log.Error().Err(err).Msgf("Cannot start the provider %T", prd)
		return
	}
}