consumer_controller.go

package controller

import (
	"context"

	"github.com/api7/api7-ingress-controller/api/v1alpha1"
	"github.com/api7/api7-ingress-controller/internal/controller/indexer"
	"github.com/api7/api7-ingress-controller/internal/provider"
	"github.com/go-logr/logr"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/event"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
)

// ConsumerReconciler  reconciles a Gateway object.
type ConsumerReconciler struct { //nolint:revive
	client.Client
	Scheme *runtime.Scheme
	Log    logr.Logger

	Provider provider.Provider
}

// SetupWithManager sets up the controller with the Manager.
func (r *ConsumerReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&v1alpha1.Consumer{},
			builder.WithPredicates(
				predicate.NewPredicateFuncs(r.checkGatewayRef),
			),
		).
		WithEventFilter(predicate.GenerationChangedPredicate{}).
		Watches(&gatewayv1.Gateway{},
			handler.EnqueueRequestsFromMapFunc(r.listConsumersForGateway),
			builder.WithPredicates(
				predicate.Funcs{
					GenericFunc: func(e event.GenericEvent) bool {
						return false
					},
					DeleteFunc: func(e event.DeleteEvent) bool {
						return false
					},
					CreateFunc: func(e event.CreateEvent) bool {
						return true
					},
					UpdateFunc: func(e event.UpdateEvent) bool {
						return true
					},
				},
			),
		).
		Watches(&corev1.Secret{},
			handler.EnqueueRequestsFromMapFunc(r.listConsumersForSecret),
		).
		Complete(r)
}

func (r *ConsumerReconciler) listConsumersForSecret(ctx context.Context, obj client.Object) []reconcile.Request {
	secret, ok := obj.(*corev1.Secret)
	if !ok {
		r.Log.Error(nil, "failed to convert to Secret", "object", obj)
		return nil
	}
	consumerList := &v1alpha1.ConsumerList{}
	if err := r.List(ctx, consumerList, client.MatchingFields{
		indexer.SecretIndexRef: indexer.GenIndexKey(secret.GetNamespace(), secret.GetName()),
	}); err != nil {
		r.Log.Error(err, "failed to list consumers")
		return nil
	}
	requests := make([]reconcile.Request, 0, len(consumerList.Items))
	for _, consumer := range consumerList.Items {
		requests = append(requests, reconcile.Request{
			NamespacedName: client.ObjectKey{
				Name:      consumer.Name,
				Namespace: consumer.Namespace,
			},
		})
	}
	return requests
}

func (r *ConsumerReconciler) listConsumersForGateway(ctx context.Context, obj client.Object) []reconcile.Request {
	gateway, ok := obj.(*gatewayv1.Gateway)
	if !ok {
		r.Log.Error(nil, "failed to convert to Gateway", "object", obj)
		return nil
	}
	consumerList := &v1alpha1.ConsumerList{}
	if err := r.List(ctx, consumerList, client.MatchingFields{
		indexer.ConsumerGatewayRef: indexer.GenIndexKey(gateway.GetNamespace(), gateway.GetName()),
	}); err != nil {
		r.Log.Error(err, "failed to list consumers")
		return nil
	}
	requests := make([]reconcile.Request, 0, len(consumerList.Items))
	for _, consumer := range consumerList.Items {
		requests = append(requests, reconcile.Request{
			NamespacedName: client.ObjectKey{
				Name:      consumer.Name,
				Namespace: consumer.Namespace,
			},
		})
	}
	return requests
}

func (r *ConsumerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	consumer := new(v1alpha1.Consumer)
	if err := r.Get(ctx, req.NamespacedName, consumer); err != nil {
		if client.IgnoreNotFound(err) == nil {
			consumer.Namespace = req.Namespace
			consumer.Name = req.Name

			consumer.TypeMeta = metav1.TypeMeta{
				Kind:       "Consumer",
				APIVersion: v1alpha1.GroupVersion.String(),
			}

			if err := r.Provider.Delete(ctx, consumer); err != nil {
				return ctrl.Result{}, err
			}
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, err
	}

	var statusErr error
	tctx := provider.NewDefaultTranslateContext()

	if err := r.processSpec(ctx, tctx, consumer); err != nil {
		r.Log.Error(err, "failed to process consumer spec", "consumer", consumer)
		statusErr = err
	}

	if err := r.Provider.Update(ctx, tctx, consumer); err != nil {
		r.Log.Error(err, "failed to update consumer", "consumer", consumer)
		statusErr = err
	}

	if err := r.updateStatus(ctx, consumer, statusErr); err != nil {
		return ctrl.Result{}, err
	}

	return ctrl.Result{}, nil
}

func (r *ConsumerReconciler) processSpec(ctx context.Context, tctx *provider.TranslateContext, consumer *v1alpha1.Consumer) error {
	for _, credential := range consumer.Spec.Credentials {
		if credential.SecretRef == nil {
			continue
		}
		ns := consumer.GetNamespace()
		if credential.SecretRef.Namespace != nil {
			ns = *credential.SecretRef.Namespace
		}
		secret := corev1.Secret{}
		if err := r.Get(ctx, client.ObjectKey{
			Name:      credential.SecretRef.Name,
			Namespace: ns,
		}, &secret); err != nil {
			if client.IgnoreNotFound(err) == nil {
				continue
			}
			r.Log.Error(err, "failed to get secret", "secret", credential.SecretRef.Name)
			return err
		}

		tctx.Secrets[types.NamespacedName{
			Namespace: ns,
			Name:      credential.SecretRef.Name,
		}] = &secret

	}
	return nil
}

func (r *ConsumerReconciler) updateStatus(ctx context.Context, consumer *v1alpha1.Consumer, err error) error {
	condition := NewCondition(consumer.Generation, true, "Successfully")
	if err != nil {
		condition = NewCondition(consumer.Generation, false, err.Error())
	}
	if !VerifyConditions(&consumer.Status.Conditions, condition) {
		return nil
	}
	meta.SetStatusCondition(&consumer.Status.Conditions, condition)
	if err := r.Status().Update(ctx, consumer); err != nil {
		r.Log.Error(err, "failed to update consumer status", "consumer", consumer)
		return err
	}
	return nil
}

func (r *ConsumerReconciler) checkGatewayRef(object client.Object) bool {
	consumer, ok := object.(*v1alpha1.Consumer)
	if !ok {
		return false
	}
	if consumer.Spec.GatewayRef.Name == "" {
		return false
	}
	if consumer.Spec.GatewayRef.Kind != nil && *consumer.Spec.GatewayRef.Kind != KindGateway {
		return false
	}
	if consumer.Spec.GatewayRef.Group != nil && *consumer.Spec.GatewayRef.Group != gatewayv1.GroupName {
		return false
	}
	ns := consumer.GetNamespace()
	if consumer.Spec.GatewayRef.Namespace != nil {
		ns = *consumer.Spec.GatewayRef.Namespace
	}
	gateway := &gatewayv1.Gateway{}
	if err := r.Get(context.Background(), client.ObjectKey{
		Name:      consumer.Spec.GatewayRef.Name,
		Namespace: ns,
	}, gateway); err != nil {
		r.Log.Error(err, "failed to get gateway", "gateway", consumer.Spec.GatewayRef.Name)
		return false
	}
	gatewayClass := &gatewayv1.GatewayClass{}
	if err := r.Client.Get(context.Background(), client.ObjectKey{Name: string(gateway.Spec.GatewayClassName)}, gatewayClass); err != nil {
		r.Log.Error(err, "failed to get gateway class", "gateway", gateway.GetName(), "gatewayclass", gateway.Spec.GatewayClassName)
		return false
	}
	return matchesController(string(gatewayClass.Spec.ControllerName))
}