prometheus/discovery/kubernetes/ingress.go

// Copyright 2016 The Prometheus 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 kubernetes

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/common/model"
	v1 "k8s.io/api/networking/v1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"

	"github.com/prometheus/prometheus/discovery/targetgroup"
)

// Ingress implements discovery of Kubernetes ingress.
type Ingress struct {
	logger   *slog.Logger
	informer cache.SharedInformer
	store    cache.Store
	queue    *workqueue.Type
}

// NewIngress returns a new ingress discovery.
func NewIngress(l *slog.Logger, inf cache.SharedInformer, eventCount *prometheus.CounterVec) *Ingress {
	ingressAddCount := eventCount.WithLabelValues(RoleIngress.String(), MetricLabelRoleAdd)
	ingressUpdateCount := eventCount.WithLabelValues(RoleIngress.String(), MetricLabelRoleUpdate)
	ingressDeleteCount := eventCount.WithLabelValues(RoleIngress.String(), MetricLabelRoleDelete)

	s := &Ingress{
		logger:   l,
		informer: inf,
		store:    inf.GetStore(),
		queue:    workqueue.NewNamed(RoleIngress.String()),
	}

	_, err := s.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(o interface{}) {
			ingressAddCount.Inc()
			s.enqueue(o)
		},
		DeleteFunc: func(o interface{}) {
			ingressDeleteCount.Inc()
			s.enqueue(o)
		},
		UpdateFunc: func(_, o interface{}) {
			ingressUpdateCount.Inc()
			s.enqueue(o)
		},
	})
	if err != nil {
		l.Error("Error adding ingresses event handler.", "err", err)
	}
	return s
}

func (i *Ingress) enqueue(obj interface{}) {
	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
	if err != nil {
		return
	}

	i.queue.Add(key)
}

// Run implements the Discoverer interface.
func (i *Ingress) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
	defer i.queue.ShutDown()

	if !cache.WaitForCacheSync(ctx.Done(), i.informer.HasSynced) {
		if !errors.Is(ctx.Err(), context.Canceled) {
			i.logger.Error("ingress informer unable to sync cache")
		}
		return
	}

	go func() {
		for i.process(ctx, ch) {
		}
	}()

	// Block until the target provider is explicitly canceled.
	<-ctx.Done()
}

func (i *Ingress) process(ctx context.Context, ch chan<- []*targetgroup.Group) bool {
	keyObj, quit := i.queue.Get()
	if quit {
		return false
	}
	defer i.queue.Done(keyObj)
	key := keyObj.(string)

	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		return true
	}

	o, exists, err := i.store.GetByKey(key)
	if err != nil {
		return true
	}
	if !exists {
		send(ctx, ch, &targetgroup.Group{Source: ingressSourceFromNamespaceAndName(namespace, name)})
		return true
	}

	var ia ingressAdaptor
	switch ingress := o.(type) {
	case *v1.Ingress:
		ia = newIngressAdaptorFromV1(ingress)
	default:
		i.logger.Error("converting to Ingress object failed", "err",
			fmt.Errorf("received unexpected object: %v", o))
		return true
	}
	send(ctx, ch, i.buildIngress(ia))
	return true
}

func ingressSource(s ingressAdaptor) string {
	return ingressSourceFromNamespaceAndName(s.namespace(), s.name())
}

func ingressSourceFromNamespaceAndName(namespace, name string) string {
	return "ingress/" + namespace + "/" + name
}

const (
	ingressSchemeLabel    = metaLabelPrefix + "ingress_scheme"
	ingressHostLabel      = metaLabelPrefix + "ingress_host"
	ingressPathLabel      = metaLabelPrefix + "ingress_path"
	ingressClassNameLabel = metaLabelPrefix + "ingress_class_name"
)

func ingressLabels(ingress ingressAdaptor) model.LabelSet {
	// Each label and annotation will create two key-value pairs in the map.
	ls := make(model.LabelSet)
	ls[namespaceLabel] = lv(ingress.namespace())
	if cls := ingress.ingressClassName(); cls != nil {
		ls[ingressClassNameLabel] = lv(*cls)
	}

	addObjectMetaLabels(ls, ingress.getObjectMeta(), RoleIngress)

	return ls
}

func pathsFromIngressPaths(ingressPaths []string) []string {
	if ingressPaths == nil {
		return []string{"/"}
	}
	paths := make([]string, len(ingressPaths))
	for n, p := range ingressPaths {
		path := p
		if p == "" {
			path = "/"
		}
		paths[n] = path
	}
	return paths
}

func (i *Ingress) buildIngress(ingress ingressAdaptor) *targetgroup.Group {
	tg := &targetgroup.Group{
		Source: ingressSource(ingress),
	}
	tg.Labels = ingressLabels(ingress)

	for _, rule := range ingress.rules() {
		scheme := "http"
		paths := pathsFromIngressPaths(rule.paths())

	out:
		for _, pattern := range ingress.tlsHosts() {
			if matchesHostnamePattern(pattern, rule.host()) {
				scheme = "https"
				break out
			}
		}

		for _, path := range paths {
			tg.Targets = append(tg.Targets, model.LabelSet{
				model.AddressLabel: lv(rule.host()),
				ingressSchemeLabel: lv(scheme),
				ingressHostLabel:   lv(rule.host()),
				ingressPathLabel:   lv(path),
			})
		}
	}

	return tg
}

// matchesHostnamePattern returns true if the host matches a wildcard DNS
// pattern or pattern and host are equal.
func matchesHostnamePattern(pattern, host string) bool {
	if pattern == host {
		return true
	}

	patternParts := strings.Split(pattern, ".")
	hostParts := strings.Split(host, ".")

	// If the first element of the pattern is not a wildcard, give up.
	if len(patternParts) == 0 || patternParts[0] != "*" {
		return false
	}

	// A wildcard match require the pattern to have the same length as the host
	// path.
	if len(patternParts) != len(hostParts) {
		return false
	}

	for i := 1; i < len(patternParts); i++ {
		if patternParts[i] != hostParts[i] {
			return false
		}
	}

	return true
}