baalajimaestro
dc5b82f6fd
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
Signed-off-by: baalajimaestro <me@baalajimaestro.me>
789 lines
25 KiB
Go
789 lines
25 KiB
Go
package ingress
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"errors"
|
|
"fmt"
|
|
"math"
|
|
"net"
|
|
"os"
|
|
"regexp"
|
|
"slices"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/cenkalti/backoff/v4"
|
|
"github.com/mitchellh/hashstructure"
|
|
"github.com/rs/zerolog/log"
|
|
ptypes "github.com/traefik/paerser/types"
|
|
"github.com/traefik/traefik/v3/pkg/config/dynamic"
|
|
"github.com/traefik/traefik/v3/pkg/job"
|
|
"github.com/traefik/traefik/v3/pkg/logs"
|
|
"github.com/traefik/traefik/v3/pkg/provider"
|
|
"github.com/traefik/traefik/v3/pkg/provider/kubernetes/k8s"
|
|
"github.com/traefik/traefik/v3/pkg/safe"
|
|
"github.com/traefik/traefik/v3/pkg/tls"
|
|
"github.com/traefik/traefik/v3/pkg/types"
|
|
corev1 "k8s.io/api/core/v1"
|
|
netv1 "k8s.io/api/networking/v1"
|
|
"k8s.io/apimachinery/pkg/labels"
|
|
)
|
|
|
|
const (
|
|
annotationKubernetesIngressClass = "kubernetes.io/ingress.class"
|
|
traefikDefaultIngressClass = "traefik"
|
|
traefikDefaultIngressClassController = "traefik.io/ingress-controller"
|
|
defaultPathMatcher = "PathPrefix"
|
|
)
|
|
|
|
// Provider holds configurations of the provider.
|
|
type Provider struct {
|
|
Endpoint string `description:"Kubernetes server endpoint (required for external cluster client)." json:"endpoint,omitempty" toml:"endpoint,omitempty" yaml:"endpoint,omitempty"`
|
|
Token types.FileOrContent `description:"Kubernetes bearer token (not needed for in-cluster client). It accepts either a token value or a file path to the token." json:"token,omitempty" toml:"token,omitempty" yaml:"token,omitempty" loggable:"false"`
|
|
CertAuthFilePath string `description:"Kubernetes certificate authority file path (not needed for in-cluster client)." json:"certAuthFilePath,omitempty" toml:"certAuthFilePath,omitempty" yaml:"certAuthFilePath,omitempty"`
|
|
Namespaces []string `description:"Kubernetes namespaces." json:"namespaces,omitempty" toml:"namespaces,omitempty" yaml:"namespaces,omitempty" export:"true"`
|
|
LabelSelector string `description:"Kubernetes Ingress label selector to use." json:"labelSelector,omitempty" toml:"labelSelector,omitempty" yaml:"labelSelector,omitempty" export:"true"`
|
|
IngressClass string `description:"Value of kubernetes.io/ingress.class annotation or IngressClass name to watch for." json:"ingressClass,omitempty" toml:"ingressClass,omitempty" yaml:"ingressClass,omitempty" export:"true"`
|
|
IngressEndpoint *EndpointIngress `description:"Kubernetes Ingress Endpoint." json:"ingressEndpoint,omitempty" toml:"ingressEndpoint,omitempty" yaml:"ingressEndpoint,omitempty" export:"true"`
|
|
ThrottleDuration ptypes.Duration `description:"Ingress refresh throttle duration" json:"throttleDuration,omitempty" toml:"throttleDuration,omitempty" yaml:"throttleDuration,omitempty" export:"true"`
|
|
AllowEmptyServices bool `description:"Allow creation of services without endpoints." json:"allowEmptyServices,omitempty" toml:"allowEmptyServices,omitempty" yaml:"allowEmptyServices,omitempty" export:"true"`
|
|
AllowExternalNameServices bool `description:"Allow ExternalName services." json:"allowExternalNameServices,omitempty" toml:"allowExternalNameServices,omitempty" yaml:"allowExternalNameServices,omitempty" export:"true"`
|
|
DisableIngressClassLookup bool `description:"Disables the lookup of IngressClasses." json:"disableIngressClassLookup,omitempty" toml:"disableIngressClassLookup,omitempty" yaml:"disableIngressClassLookup,omitempty" export:"true"`
|
|
|
|
lastConfiguration safe.Safe
|
|
|
|
routerTransform k8s.RouterTransform
|
|
}
|
|
|
|
func (p *Provider) SetRouterTransform(routerTransform k8s.RouterTransform) {
|
|
p.routerTransform = routerTransform
|
|
}
|
|
|
|
func (p *Provider) applyRouterTransform(ctx context.Context, rt *dynamic.Router, ingress *netv1.Ingress) {
|
|
if p.routerTransform == nil {
|
|
return
|
|
}
|
|
|
|
err := p.routerTransform.Apply(ctx, rt, ingress)
|
|
if err != nil {
|
|
log.Ctx(ctx).Error().Err(err).Msg("Apply router transform")
|
|
}
|
|
}
|
|
|
|
// EndpointIngress holds the endpoint information for the Kubernetes provider.
|
|
type EndpointIngress struct {
|
|
IP string `description:"IP used for Kubernetes Ingress endpoints." json:"ip,omitempty" toml:"ip,omitempty" yaml:"ip,omitempty"`
|
|
Hostname string `description:"Hostname used for Kubernetes Ingress endpoints." json:"hostname,omitempty" toml:"hostname,omitempty" yaml:"hostname,omitempty"`
|
|
PublishedService string `description:"Published Kubernetes Service to copy status from." json:"publishedService,omitempty" toml:"publishedService,omitempty" yaml:"publishedService,omitempty"`
|
|
}
|
|
|
|
func (p *Provider) newK8sClient(ctx context.Context) (*clientWrapper, error) {
|
|
_, err := labels.Parse(p.LabelSelector)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid ingress label selector: %q", p.LabelSelector)
|
|
}
|
|
|
|
logger := log.Ctx(ctx)
|
|
|
|
logger.Info().Msgf("ingress label selector is: %q", p.LabelSelector)
|
|
|
|
withEndpoint := ""
|
|
if p.Endpoint != "" {
|
|
withEndpoint = fmt.Sprintf(" with endpoint %v", p.Endpoint)
|
|
}
|
|
|
|
var cl *clientWrapper
|
|
switch {
|
|
case os.Getenv("KUBERNETES_SERVICE_HOST") != "" && os.Getenv("KUBERNETES_SERVICE_PORT") != "":
|
|
logger.Info().Msgf("Creating in-cluster Provider client%s", withEndpoint)
|
|
cl, err = newInClusterClient(p.Endpoint)
|
|
case os.Getenv("KUBECONFIG") != "":
|
|
logger.Info().Msgf("Creating cluster-external Provider client from KUBECONFIG %s", os.Getenv("KUBECONFIG"))
|
|
cl, err = newExternalClusterClientFromFile(os.Getenv("KUBECONFIG"))
|
|
default:
|
|
logger.Info().Msgf("Creating cluster-external Provider client%s", withEndpoint)
|
|
cl, err = newExternalClusterClient(p.Endpoint, p.CertAuthFilePath, p.Token)
|
|
}
|
|
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cl.ingressLabelSelector = p.LabelSelector
|
|
cl.disableIngressClassInformer = p.DisableIngressClassLookup
|
|
return cl, nil
|
|
}
|
|
|
|
// Init the provider.
|
|
func (p *Provider) Init() error {
|
|
return nil
|
|
}
|
|
|
|
// Provide allows the k8s provider to provide configurations to traefik
|
|
// using the given configuration channel.
|
|
func (p *Provider) Provide(configurationChan chan<- dynamic.Message, pool *safe.Pool) error {
|
|
logger := log.With().Str(logs.ProviderName, "kubernetes").Logger()
|
|
ctxLog := logger.WithContext(context.Background())
|
|
|
|
k8sClient, err := p.newK8sClient(ctxLog)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if p.AllowExternalNameServices {
|
|
logger.Warn().Msg("ExternalName service loading is enabled, please ensure that this is expected (see AllowExternalNameServices option)")
|
|
}
|
|
|
|
pool.GoCtx(func(ctxPool context.Context) {
|
|
operation := func() error {
|
|
eventsChan, err := k8sClient.WatchAll(p.Namespaces, ctxPool.Done())
|
|
if err != nil {
|
|
logger.Error().Err(err).Msg("Error watching kubernetes events")
|
|
timer := time.NewTimer(1 * time.Second)
|
|
select {
|
|
case <-timer.C:
|
|
return err
|
|
case <-ctxPool.Done():
|
|
return nil
|
|
}
|
|
}
|
|
|
|
throttleDuration := time.Duration(p.ThrottleDuration)
|
|
throttledChan := throttleEvents(ctxLog, throttleDuration, pool, eventsChan)
|
|
if throttledChan != nil {
|
|
eventsChan = throttledChan
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case <-ctxPool.Done():
|
|
return nil
|
|
case event := <-eventsChan:
|
|
// Note that event is the *first* event that came in during this
|
|
// throttling interval -- if we're hitting our throttle, we may have
|
|
// dropped events. This is fine, because we don't treat different
|
|
// event types differently. But if we do in the future, we'll need to
|
|
// track more information about the dropped events.
|
|
conf := p.loadConfigurationFromIngresses(ctxLog, k8sClient)
|
|
|
|
confHash, err := hashstructure.Hash(conf, nil)
|
|
switch {
|
|
case err != nil:
|
|
logger.Error().Msg("Unable to hash the configuration")
|
|
case p.lastConfiguration.Get() == confHash:
|
|
logger.Debug().Msgf("Skipping Kubernetes event kind %T", event)
|
|
default:
|
|
p.lastConfiguration.Set(confHash)
|
|
configurationChan <- dynamic.Message{
|
|
ProviderName: "kubernetes",
|
|
Configuration: conf,
|
|
}
|
|
}
|
|
|
|
// If we're throttling, we sleep here for the throttle duration to
|
|
// enforce that we don't refresh faster than our throttle. time.Sleep
|
|
// returns immediately if p.ThrottleDuration is 0 (no throttle).
|
|
time.Sleep(throttleDuration)
|
|
}
|
|
}
|
|
}
|
|
|
|
notify := func(err error, time time.Duration) {
|
|
logger.Error().Err(err).Msgf("Provider error, retrying in %s", time)
|
|
}
|
|
|
|
err := backoff.RetryNotify(safe.OperationWithRecover(operation), backoff.WithContext(job.NewBackOff(backoff.NewExponentialBackOff()), ctxPool), notify)
|
|
if err != nil {
|
|
logger.Error().Err(err).Msg("Cannot retrieve data")
|
|
}
|
|
})
|
|
|
|
return nil
|
|
}
|
|
|
|
func (p *Provider) loadConfigurationFromIngresses(ctx context.Context, client Client) *dynamic.Configuration {
|
|
conf := &dynamic.Configuration{
|
|
HTTP: &dynamic.HTTPConfiguration{
|
|
Routers: map[string]*dynamic.Router{},
|
|
Middlewares: map[string]*dynamic.Middleware{},
|
|
Services: map[string]*dynamic.Service{},
|
|
},
|
|
TCP: &dynamic.TCPConfiguration{},
|
|
}
|
|
|
|
var ingressClasses []*netv1.IngressClass
|
|
|
|
if !p.DisableIngressClassLookup {
|
|
ics, err := client.GetIngressClasses()
|
|
if err != nil {
|
|
log.Ctx(ctx).Warn().Err(err).Msg("Failed to list ingress classes")
|
|
}
|
|
|
|
if p.IngressClass != "" {
|
|
ingressClasses = filterIngressClassByName(p.IngressClass, ics)
|
|
} else {
|
|
ingressClasses = ics
|
|
}
|
|
}
|
|
|
|
ingresses := client.GetIngresses()
|
|
|
|
certConfigs := make(map[string]*tls.CertAndStores)
|
|
for _, ingress := range ingresses {
|
|
logger := log.Ctx(ctx).With().Str("ingress", ingress.Name).Str("namespace", ingress.Namespace).Logger()
|
|
ctx = logger.WithContext(ctx)
|
|
|
|
if !p.shouldProcessIngress(ingress, ingressClasses) {
|
|
continue
|
|
}
|
|
|
|
rtConfig, err := parseRouterConfig(ingress.Annotations)
|
|
if err != nil {
|
|
logger.Error().Err(err).Msg("Failed to parse annotations")
|
|
continue
|
|
}
|
|
|
|
err = getCertificates(ctx, ingress, client, certConfigs)
|
|
if err != nil {
|
|
logger.Error().Err(err).Msg("Error configuring TLS")
|
|
}
|
|
|
|
if len(ingress.Spec.Rules) == 0 && ingress.Spec.DefaultBackend != nil {
|
|
if _, ok := conf.HTTP.Services["default-backend"]; ok {
|
|
logger.Error().Msg("The default backend already exists.")
|
|
continue
|
|
}
|
|
|
|
service, err := p.loadService(client, ingress.Namespace, *ingress.Spec.DefaultBackend)
|
|
if err != nil {
|
|
logger.Error().
|
|
Str("serviceName", ingress.Spec.DefaultBackend.Service.Name).
|
|
Str("servicePort", ingress.Spec.DefaultBackend.Service.Port.String()).
|
|
Err(err).
|
|
Msg("Cannot create service")
|
|
continue
|
|
}
|
|
|
|
if len(service.LoadBalancer.Servers) == 0 && !p.AllowEmptyServices {
|
|
logger.Error().
|
|
Str("serviceName", ingress.Spec.DefaultBackend.Service.Name).
|
|
Str("servicePort", ingress.Spec.DefaultBackend.Service.Port.String()).
|
|
Msg("Skipping service: no endpoints found")
|
|
continue
|
|
}
|
|
|
|
rt := &dynamic.Router{
|
|
Rule: "PathPrefix(`/`)",
|
|
Priority: math.MinInt32,
|
|
Service: "default-backend",
|
|
}
|
|
|
|
if rtConfig != nil && rtConfig.Router != nil {
|
|
rt.EntryPoints = rtConfig.Router.EntryPoints
|
|
rt.Middlewares = rtConfig.Router.Middlewares
|
|
rt.TLS = rtConfig.Router.TLS
|
|
}
|
|
|
|
p.applyRouterTransform(ctx, rt, ingress)
|
|
|
|
conf.HTTP.Routers["default-router"] = rt
|
|
conf.HTTP.Services["default-backend"] = service
|
|
}
|
|
|
|
routers := map[string][]*dynamic.Router{}
|
|
|
|
for _, rule := range ingress.Spec.Rules {
|
|
if err := p.updateIngressStatus(ingress, client); err != nil {
|
|
logger.Error().Err(err).Msg("Error while updating ingress status")
|
|
}
|
|
|
|
if rule.HTTP == nil {
|
|
continue
|
|
}
|
|
|
|
for _, pa := range rule.HTTP.Paths {
|
|
service, err := p.loadService(client, ingress.Namespace, pa.Backend)
|
|
if err != nil {
|
|
logger.Error().
|
|
Str("serviceName", pa.Backend.Service.Name).
|
|
Str("servicePort", pa.Backend.Service.Port.String()).
|
|
Err(err).
|
|
Msg("Cannot create service")
|
|
continue
|
|
}
|
|
|
|
if len(service.LoadBalancer.Servers) == 0 && !p.AllowEmptyServices {
|
|
logger.Error().
|
|
Str("serviceName", pa.Backend.Service.Name).
|
|
Str("servicePort", pa.Backend.Service.Port.String()).
|
|
Msg("Skipping service: no endpoints found")
|
|
continue
|
|
}
|
|
|
|
portString := pa.Backend.Service.Port.Name
|
|
|
|
if len(pa.Backend.Service.Port.Name) == 0 {
|
|
portString = strconv.Itoa(int(pa.Backend.Service.Port.Number))
|
|
}
|
|
|
|
serviceName := provider.Normalize(ingress.Namespace + "-" + pa.Backend.Service.Name + "-" + portString)
|
|
conf.HTTP.Services[serviceName] = service
|
|
|
|
rt := loadRouter(rule, pa, rtConfig, serviceName)
|
|
|
|
p.applyRouterTransform(ctx, rt, ingress)
|
|
|
|
routerKey := strings.TrimPrefix(provider.Normalize(ingress.Namespace+"-"+ingress.Name+"-"+rule.Host+pa.Path), "-")
|
|
|
|
routers[routerKey] = append(routers[routerKey], rt)
|
|
}
|
|
}
|
|
|
|
for routerKey, conflictingRouters := range routers {
|
|
if len(conflictingRouters) == 1 {
|
|
conf.HTTP.Routers[routerKey] = conflictingRouters[0]
|
|
continue
|
|
}
|
|
|
|
logger.Debug().Msgf("Multiple routers are defined with the same key %q, generating hashes to avoid conflicts", routerKey)
|
|
|
|
for _, router := range conflictingRouters {
|
|
key, err := makeRouterKeyWithHash(routerKey, router.Rule)
|
|
if err != nil {
|
|
logger.Error().Err(err).Send()
|
|
continue
|
|
}
|
|
|
|
conf.HTTP.Routers[key] = router
|
|
}
|
|
}
|
|
}
|
|
|
|
certs := getTLSConfig(certConfigs)
|
|
if len(certs) > 0 {
|
|
conf.TLS = &dynamic.TLSConfiguration{
|
|
Certificates: certs,
|
|
}
|
|
}
|
|
|
|
return conf
|
|
}
|
|
|
|
func (p *Provider) updateIngressStatus(ing *netv1.Ingress, k8sClient Client) error {
|
|
// Only process if an EndpointIngress has been configured.
|
|
if p.IngressEndpoint == nil {
|
|
return nil
|
|
}
|
|
|
|
if len(p.IngressEndpoint.PublishedService) == 0 {
|
|
if len(p.IngressEndpoint.IP) == 0 && len(p.IngressEndpoint.Hostname) == 0 {
|
|
return errors.New("publishedService or ip or hostname must be defined")
|
|
}
|
|
|
|
return k8sClient.UpdateIngressStatus(ing, []netv1.IngressLoadBalancerIngress{{IP: p.IngressEndpoint.IP, Hostname: p.IngressEndpoint.Hostname}})
|
|
}
|
|
|
|
serviceInfo := strings.Split(p.IngressEndpoint.PublishedService, "/")
|
|
if len(serviceInfo) != 2 {
|
|
return fmt.Errorf("invalid publishedService format (expected 'namespace/service' format): %s", p.IngressEndpoint.PublishedService)
|
|
}
|
|
|
|
serviceNamespace, serviceName := serviceInfo[0], serviceInfo[1]
|
|
|
|
service, exists, err := k8sClient.GetService(serviceNamespace, serviceName)
|
|
if err != nil {
|
|
return fmt.Errorf("cannot get service %s, received error: %w", p.IngressEndpoint.PublishedService, err)
|
|
}
|
|
|
|
if exists && service.Status.LoadBalancer.Ingress == nil {
|
|
// service exists, but has no Load Balancer status
|
|
log.Debug().Msgf("Skipping updating Ingress %s/%s due to service %s having no status set", ing.Namespace, ing.Name, p.IngressEndpoint.PublishedService)
|
|
return nil
|
|
}
|
|
|
|
if !exists {
|
|
return fmt.Errorf("missing service: %s", p.IngressEndpoint.PublishedService)
|
|
}
|
|
|
|
ingresses, err := convertSlice[netv1.IngressLoadBalancerIngress](service.Status.LoadBalancer.Ingress)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return k8sClient.UpdateIngressStatus(ing, ingresses)
|
|
}
|
|
|
|
func (p *Provider) shouldProcessIngress(ingress *netv1.Ingress, ingressClasses []*netv1.IngressClass) bool {
|
|
// configuration through the new kubernetes ingressClass
|
|
if ingress.Spec.IngressClassName != nil {
|
|
return slices.ContainsFunc(ingressClasses, func(ic *netv1.IngressClass) bool {
|
|
return *ingress.Spec.IngressClassName == ic.ObjectMeta.Name
|
|
})
|
|
}
|
|
|
|
return p.IngressClass == ingress.Annotations[annotationKubernetesIngressClass] ||
|
|
len(p.IngressClass) == 0 && ingress.Annotations[annotationKubernetesIngressClass] == traefikDefaultIngressClass
|
|
}
|
|
|
|
func buildHostRule(host string) string {
|
|
if strings.HasPrefix(host, "*.") {
|
|
host = strings.Replace(regexp.QuoteMeta(host), `\*\.`, `[a-zA-Z0-9-]+\.`, 1)
|
|
return fmt.Sprintf("HostRegexp(`^%s$`)", host)
|
|
}
|
|
|
|
return fmt.Sprintf("Host(`%s`)", host)
|
|
}
|
|
|
|
func getCertificates(ctx context.Context, ingress *netv1.Ingress, k8sClient Client, tlsConfigs map[string]*tls.CertAndStores) error {
|
|
for _, t := range ingress.Spec.TLS {
|
|
if t.SecretName == "" {
|
|
log.Ctx(ctx).Debug().Msg("Skipping TLS sub-section: No secret name provided")
|
|
continue
|
|
}
|
|
|
|
configKey := ingress.Namespace + "-" + t.SecretName
|
|
if _, tlsExists := tlsConfigs[configKey]; !tlsExists {
|
|
secret, exists, err := k8sClient.GetSecret(ingress.Namespace, t.SecretName)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to fetch secret %s/%s: %w", ingress.Namespace, t.SecretName, err)
|
|
}
|
|
if !exists {
|
|
return fmt.Errorf("secret %s/%s does not exist", ingress.Namespace, t.SecretName)
|
|
}
|
|
|
|
cert, key, err := getCertificateBlocks(secret, ingress.Namespace, t.SecretName)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
tlsConfigs[configKey] = &tls.CertAndStores{
|
|
Certificate: tls.Certificate{
|
|
CertFile: types.FileOrContent(cert),
|
|
KeyFile: types.FileOrContent(key),
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func getCertificateBlocks(secret *corev1.Secret, namespace, secretName string) (string, string, error) {
|
|
var missingEntries []string
|
|
|
|
tlsCrtData, tlsCrtExists := secret.Data["tls.crt"]
|
|
if !tlsCrtExists {
|
|
missingEntries = append(missingEntries, "tls.crt")
|
|
}
|
|
|
|
tlsKeyData, tlsKeyExists := secret.Data["tls.key"]
|
|
if !tlsKeyExists {
|
|
missingEntries = append(missingEntries, "tls.key")
|
|
}
|
|
|
|
if len(missingEntries) > 0 {
|
|
return "", "", fmt.Errorf("secret %s/%s is missing the following TLS data entries: %s",
|
|
namespace, secretName, strings.Join(missingEntries, ", "))
|
|
}
|
|
|
|
cert := string(tlsCrtData)
|
|
if cert == "" {
|
|
missingEntries = append(missingEntries, "tls.crt")
|
|
}
|
|
|
|
key := string(tlsKeyData)
|
|
if key == "" {
|
|
missingEntries = append(missingEntries, "tls.key")
|
|
}
|
|
|
|
if len(missingEntries) > 0 {
|
|
return "", "", fmt.Errorf("secret %s/%s contains the following empty TLS data entries: %s",
|
|
namespace, secretName, strings.Join(missingEntries, ", "))
|
|
}
|
|
|
|
return cert, key, nil
|
|
}
|
|
|
|
func getTLSConfig(tlsConfigs map[string]*tls.CertAndStores) []*tls.CertAndStores {
|
|
var secretNames []string
|
|
for secretName := range tlsConfigs {
|
|
secretNames = append(secretNames, secretName)
|
|
}
|
|
sort.Strings(secretNames)
|
|
|
|
var configs []*tls.CertAndStores
|
|
for _, secretName := range secretNames {
|
|
configs = append(configs, tlsConfigs[secretName])
|
|
}
|
|
|
|
return configs
|
|
}
|
|
|
|
func (p *Provider) loadService(client Client, namespace string, backend netv1.IngressBackend) (*dynamic.Service, error) {
|
|
if backend.Resource != nil {
|
|
// https://kubernetes.io/docs/concepts/services-networking/ingress/#resource-backend
|
|
return nil, errors.New("resource backends are not supported")
|
|
}
|
|
|
|
if backend.Service == nil {
|
|
return nil, errors.New("missing service definition")
|
|
}
|
|
|
|
service, exists, err := client.GetService(namespace, backend.Service.Name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if !exists {
|
|
return nil, errors.New("service not found")
|
|
}
|
|
|
|
if !p.AllowExternalNameServices && service.Spec.Type == corev1.ServiceTypeExternalName {
|
|
return nil, fmt.Errorf("externalName services not allowed: %s/%s", namespace, backend.Service.Name)
|
|
}
|
|
|
|
var portName string
|
|
var portSpec corev1.ServicePort
|
|
var match bool
|
|
for _, p := range service.Spec.Ports {
|
|
if backend.Service.Port.Number == p.Port || (backend.Service.Port.Name == p.Name && len(p.Name) > 0) {
|
|
portName = p.Name
|
|
portSpec = p
|
|
match = true
|
|
break
|
|
}
|
|
}
|
|
|
|
if !match {
|
|
return nil, errors.New("service port not found")
|
|
}
|
|
|
|
lb := &dynamic.ServersLoadBalancer{}
|
|
lb.SetDefaults()
|
|
|
|
svc := &dynamic.Service{LoadBalancer: lb}
|
|
|
|
svcConfig, err := parseServiceConfig(service.Annotations)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if svcConfig != nil && svcConfig.Service != nil {
|
|
svc.LoadBalancer.Sticky = svcConfig.Service.Sticky
|
|
|
|
if svcConfig.Service.PassHostHeader != nil {
|
|
svc.LoadBalancer.PassHostHeader = svcConfig.Service.PassHostHeader
|
|
}
|
|
|
|
if svcConfig.Service.ServersTransport != "" {
|
|
svc.LoadBalancer.ServersTransport = svcConfig.Service.ServersTransport
|
|
}
|
|
|
|
if svcConfig.Service.NativeLB {
|
|
address, err := getNativeServiceAddress(*service, portSpec)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("getting native Kubernetes Service address: %w", err)
|
|
}
|
|
|
|
protocol := getProtocol(portSpec, portSpec.Name, svcConfig)
|
|
|
|
svc.LoadBalancer.Servers = []dynamic.Server{
|
|
{URL: fmt.Sprintf("%s://%s", protocol, address)},
|
|
}
|
|
|
|
return svc, nil
|
|
}
|
|
|
|
if svcConfig.Service.NodePortLB && service.Spec.Type == corev1.ServiceTypeNodePort {
|
|
nodes, nodesExists, nodesErr := client.GetNodes()
|
|
if nodesErr != nil {
|
|
return nil, nodesErr
|
|
}
|
|
|
|
if !nodesExists || len(nodes) == 0 {
|
|
return nil, fmt.Errorf("nodes not found in namespace %s", namespace)
|
|
}
|
|
|
|
protocol := getProtocol(portSpec, portSpec.Name, svcConfig)
|
|
|
|
var servers []dynamic.Server
|
|
|
|
for _, node := range nodes {
|
|
for _, addr := range node.Status.Addresses {
|
|
if addr.Type == corev1.NodeInternalIP {
|
|
hostPort := net.JoinHostPort(addr.Address, strconv.Itoa(int(portSpec.NodePort)))
|
|
|
|
servers = append(servers, dynamic.Server{
|
|
URL: fmt.Sprintf("%s://%s", protocol, hostPort),
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
if len(servers) == 0 {
|
|
return nil, fmt.Errorf("no servers were generated for service %s in namespace", backend.Service.Name)
|
|
}
|
|
|
|
svc.LoadBalancer.Servers = servers
|
|
|
|
return svc, nil
|
|
}
|
|
}
|
|
|
|
if service.Spec.Type == corev1.ServiceTypeExternalName {
|
|
protocol := getProtocol(portSpec, portSpec.Name, svcConfig)
|
|
hostPort := net.JoinHostPort(service.Spec.ExternalName, strconv.Itoa(int(portSpec.Port)))
|
|
|
|
svc.LoadBalancer.Servers = []dynamic.Server{
|
|
{URL: fmt.Sprintf("%s://%s", protocol, hostPort)},
|
|
}
|
|
|
|
return svc, nil
|
|
}
|
|
|
|
endpoints, endpointsExists, endpointsErr := client.GetEndpoints(namespace, backend.Service.Name)
|
|
if endpointsErr != nil {
|
|
return nil, endpointsErr
|
|
}
|
|
|
|
if !endpointsExists {
|
|
return nil, errors.New("endpoints not found")
|
|
}
|
|
|
|
for _, subset := range endpoints.Subsets {
|
|
var port int32
|
|
for _, p := range subset.Ports {
|
|
if portName == p.Name {
|
|
port = p.Port
|
|
break
|
|
}
|
|
}
|
|
|
|
if port == 0 {
|
|
continue
|
|
}
|
|
|
|
protocol := getProtocol(portSpec, portName, svcConfig)
|
|
|
|
for _, addr := range subset.Addresses {
|
|
hostPort := net.JoinHostPort(addr.IP, strconv.Itoa(int(port)))
|
|
|
|
svc.LoadBalancer.Servers = append(svc.LoadBalancer.Servers, dynamic.Server{
|
|
URL: fmt.Sprintf("%s://%s", protocol, hostPort),
|
|
})
|
|
}
|
|
}
|
|
|
|
return svc, nil
|
|
}
|
|
|
|
func getNativeServiceAddress(service corev1.Service, svcPort corev1.ServicePort) (string, error) {
|
|
if service.Spec.ClusterIP == "None" {
|
|
return "", fmt.Errorf("no clusterIP on headless service: %s/%s", service.Namespace, service.Name)
|
|
}
|
|
|
|
if service.Spec.ClusterIP == "" {
|
|
return "", fmt.Errorf("no clusterIP found for service: %s/%s", service.Namespace, service.Name)
|
|
}
|
|
|
|
return net.JoinHostPort(service.Spec.ClusterIP, strconv.Itoa(int(svcPort.Port))), nil
|
|
}
|
|
|
|
func getProtocol(portSpec corev1.ServicePort, portName string, svcConfig *ServiceConfig) string {
|
|
if svcConfig != nil && svcConfig.Service != nil && svcConfig.Service.ServersScheme != "" {
|
|
return svcConfig.Service.ServersScheme
|
|
}
|
|
|
|
protocol := "http"
|
|
if portSpec.Port == 443 || strings.HasPrefix(portName, "https") {
|
|
protocol = "https"
|
|
}
|
|
|
|
return protocol
|
|
}
|
|
|
|
func makeRouterKeyWithHash(key, rule string) (string, error) {
|
|
h := sha256.New()
|
|
if _, err := h.Write([]byte(rule)); err != nil {
|
|
return "", err
|
|
}
|
|
|
|
dupKey := fmt.Sprintf("%s-%.10x", key, h.Sum(nil))
|
|
|
|
return dupKey, nil
|
|
}
|
|
|
|
func loadRouter(rule netv1.IngressRule, pa netv1.HTTPIngressPath, rtConfig *RouterConfig, serviceName string) *dynamic.Router {
|
|
var rules []string
|
|
if len(rule.Host) > 0 {
|
|
rules = []string{buildHostRule(rule.Host)}
|
|
}
|
|
|
|
if len(pa.Path) > 0 {
|
|
matcher := defaultPathMatcher
|
|
|
|
if pa.PathType == nil || *pa.PathType == "" || *pa.PathType == netv1.PathTypeImplementationSpecific {
|
|
if rtConfig != nil && rtConfig.Router != nil && rtConfig.Router.PathMatcher != "" {
|
|
matcher = rtConfig.Router.PathMatcher
|
|
}
|
|
} else if *pa.PathType == netv1.PathTypeExact {
|
|
matcher = "Path"
|
|
}
|
|
|
|
rules = append(rules, fmt.Sprintf("%s(`%s`)", matcher, pa.Path))
|
|
}
|
|
|
|
rt := &dynamic.Router{
|
|
Rule: strings.Join(rules, " && "),
|
|
Service: serviceName,
|
|
}
|
|
|
|
if rtConfig != nil && rtConfig.Router != nil {
|
|
rt.Priority = rtConfig.Router.Priority
|
|
rt.EntryPoints = rtConfig.Router.EntryPoints
|
|
rt.Middlewares = rtConfig.Router.Middlewares
|
|
|
|
if rtConfig.Router.TLS != nil {
|
|
rt.TLS = rtConfig.Router.TLS
|
|
}
|
|
}
|
|
|
|
return rt
|
|
}
|
|
|
|
func throttleEvents(ctx context.Context, throttleDuration time.Duration, pool *safe.Pool, eventsChan <-chan interface{}) chan interface{} {
|
|
if throttleDuration == 0 {
|
|
return nil
|
|
}
|
|
|
|
// Create a buffered channel to hold the pending event (if we're delaying processing the event due to throttling).
|
|
eventsChanBuffered := make(chan interface{}, 1)
|
|
|
|
// Run a goroutine that reads events from eventChan and does a
|
|
// non-blocking write to pendingEvent. This guarantees that writing to
|
|
// eventChan will never block, and that pendingEvent will have
|
|
// something in it if there's been an event since we read from that channel.
|
|
pool.GoCtx(func(ctxPool context.Context) {
|
|
for {
|
|
select {
|
|
case <-ctxPool.Done():
|
|
return
|
|
case nextEvent := <-eventsChan:
|
|
select {
|
|
case eventsChanBuffered <- nextEvent:
|
|
default:
|
|
// We already have an event in eventsChanBuffered, so we'll
|
|
// do a refresh as soon as our throttle allows us to. It's fine
|
|
// to drop the event and keep whatever's in the buffer -- we
|
|
// don't do different things for different events.
|
|
log.Ctx(ctx).Debug().Msgf("Dropping event kind %T due to throttling", nextEvent)
|
|
}
|
|
}
|
|
}
|
|
})
|
|
|
|
return eventsChanBuffered
|
|
}
|