<commit_before>/*
Copyright 2020 The HAProxy Ingress Controller 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 controller

import (
	"fmt"
	"reflect"
	"time"

	api "k8s.io/api/core/v1"
	networking "k8s.io/api/networking/v1beta1"
	"k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/client-go/informers"
	informersv1 "k8s.io/client-go/informers/core/v1"
	informersv1beta1 "k8s.io/client-go/informers/networking/v1beta1"
	k8s "k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/fake"
	listersv1 "k8s.io/client-go/listers/core/v1"
	listersv1beta1 "k8s.io/client-go/listers/networking/v1beta1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/record"

	"github.com/jcmoraisjr/haproxy-ingress/pkg/types"
)

// ListerEvents ...
type ListerEvents interface {
	IsValidIngress(ing *networking.Ingress) bool
	IsValidIngressClass(ingClass *networking.IngressClass) bool
	IsValidConfigMap(cm *api.ConfigMap) bool
	Notify(old, cur interface{})
}

type listers struct {
	events   ListerEvents
	logger   types.Logger
	recorder record.EventRecorder
	running  bool
	//
	hasPodLister  bool
	hasNodeLister bool
	//
	ingressLister      listersv1beta1.IngressLister
	ingressClassLister listersv1beta1.IngressClassLister
	endpointLister     listersv1.EndpointsLister
	serviceLister      listersv1.ServiceLister
	secretLister       listersv1.SecretLister
	configMapLister    listersv1.ConfigMapLister
	podLister          listersv1.PodLister
	nodeLister         listersv1.NodeLister
	//
	ingressInformer      cache.SharedInformer
	ingressClassInformer cache.SharedInformer
	endpointInformer     cache.SharedInformer
	serviceInformer      cache.SharedInformer
	secretInformer       cache.SharedInformer
	configMapInformer    cache.SharedInformer
	podInformer          cache.SharedInformer
	nodeInformer         cache.SharedInformer
}

func createListers(
	events ListerEvents,
	logger types.Logger,
	recorder record.EventRecorder,
	client k8s.Interface,
	watchNamespace string,
	isolateNamespace bool,
	podWatch bool,
	resync time.Duration,
) *listers {
	clusterWatch := watchNamespace == api.NamespaceAll
	clusterOption := informers.WithTweakListOptions(nil)
	namespaceOption := informers.WithNamespace(watchNamespace)
	var ingressInformer, resourceInformer, localInformer informers.SharedInformerFactory
	if clusterWatch {
		ingressInformer = informers.NewSharedInformerFactoryWithOptions(client, resync, clusterOption)
		resourceInformer = ingressInformer
	} else if isolateNamespace {
		ingressInformer = informers.NewSharedInformerFactoryWithOptions(client, resync, namespaceOption)
		resourceInformer = ingressInformer
	} else {
		ingressInformer = informers.NewSharedInformerFactoryWithOptions(client, resync, namespaceOption)
		resourceInformer = informers.NewSharedInformerFactoryWithOptions(client, resync, clusterOption)
	}
	if !podWatch || !clusterWatch {
		localInformer = informers.NewSharedInformerFactory(fake.NewSimpleClientset(), 0)
	}
	l := &listers{
		events:   events,
		recorder: recorder,
		logger:   logger,
	}
	l.createIngressLister(ingressInformer.Networking().V1beta1().Ingresses())
	l.createIngressClassLister(ingressInformer.Networking().V1beta1().IngressClasses())
	l.createEndpointLister(resourceInformer.Core().V1().Endpoints())
	l.createServiceLister(resourceInformer.Core().V1().Services())
	l.createSecretLister(resourceInformer.Core().V1().Secrets())
	l.createConfigMapLister(resourceInformer.Core().V1().ConfigMaps())
	if podWatch {
		l.createPodLister(ingressInformer.Core().V1().Pods())
		l.hasPodLister = true
	} else {
		l.createPodLister(localInformer.Core().V1().Pods())
	}
	if clusterWatch {
		// ignoring --disable-node-list
		l.createNodeLister(resourceInformer.Core().V1().Nodes())
		l.hasNodeLister = true
	} else {
		l.createNodeLister(localInformer.Core().V1().Nodes())
	}
	return l
}

func (l *listers) RunAsync(stopCh <-chan struct{}) {
	go l.ingressInformer.Run(stopCh)
	go l.ingressClassInformer.Run(stopCh)
	go l.endpointInformer.Run(stopCh)
	go l.serviceInformer.Run(stopCh)
	go l.secretInformer.Run(stopCh)
	go l.configMapInformer.Run(stopCh)
	go l.podInformer.Run(stopCh)
	go l.nodeInformer.Run(stopCh)
	l.logger.Info("loading object cache...")
	synced := cache.WaitForCacheSync(stopCh,
		l.ingressInformer.HasSynced,
		l.ingressClassInformer.HasSynced,
		l.endpointInformer.HasSynced,
		l.serviceInformer.HasSynced,
		l.secretInformer.HasSynced,
		l.configMapInformer.HasSynced,
		l.podInformer.HasSynced,
		l.nodeInformer.HasSynced,
	)
	if synced {
		l.logger.Info("cache successfully synced")
		l.running = true
	} else {
		runtime.HandleError(fmt.Errorf("initial cache sync has timed out or shutdown has requested"))
	}
}

func (l *listers) createIngressLister(informer informersv1beta1.IngressInformer) {
	l.ingressLister = informer.Lister()
	l.ingressInformer = informer.Informer()
	l.ingressInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			ing := obj.(*networking.Ingress)
			if l.events.IsValidIngress(ing) {
				l.events.Notify(nil, ing)
				if l.running {
					l.recorder.Eventf(ing, api.EventTypeNormal, "CREATE", "Ingress %s/%s", ing.Namespace, ing.Name)
				}
			}
		},
		UpdateFunc: func(old, cur interface{}) {
			if reflect.DeepEqual(old, cur) {
				return
			}
			oldIng := old.(*networking.Ingress)<commit_msg>wait ingress class initialization<commit_after>/*
Copyright 2020 The HAProxy Ingress Controller 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 controller

import (
	"fmt"
	"reflect"
	"time"

	api "k8s.io/api/core/v1"
	networking "k8s.io/api/networking/v1beta1"
	"k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/client-go/informers"
	informersv1 "k8s.io/client-go/informers/core/v1"
	informersv1beta1 "k8s.io/client-go/informers/networking/v1beta1"
	k8s "k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/fake"
	listersv1 "k8s.io/client-go/listers/core/v1"
	listersv1beta1 "k8s.io/client-go/listers/networking/v1beta1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/record"

	"github.com/jcmoraisjr/haproxy-ingress/pkg/types"
)

// ListerEvents ...
type ListerEvents interface {
	IsValidIngress(ing *networking.Ingress) bool
	IsValidIngressClass(ingClass *networking.IngressClass) bool
	IsValidConfigMap(cm *api.ConfigMap) bool
	Notify(old, cur interface{})
}

type listers struct {
	events   ListerEvents
	logger   types.Logger
	recorder record.EventRecorder
	running  bool
	//
	hasPodLister  bool
	hasNodeLister bool
	//
	ingressLister      listersv1beta1.IngressLister
	ingressClassLister listersv1beta1.IngressClassLister
	endpointLister     listersv1.EndpointsLister
	serviceLister      listersv1.ServiceLister
	secretLister       listersv1.SecretLister
	configMapLister    listersv1.ConfigMapLister
	podLister          listersv1.PodLister
	nodeLister         listersv1.NodeLister
	//
	ingressInformer      cache.SharedInformer
	ingressClassInformer cache.SharedInformer
	endpointInformer     cache.SharedInformer
	serviceInformer      cache.SharedInformer
	secretInformer       cache.SharedInformer
	configMapInformer    cache.SharedInformer
	podInformer          cache.SharedInformer
	nodeInformer         cache.SharedInformer
}

func createListers(
	events ListerEvents,
	logger types.Logger,
	recorder record.EventRecorder,
	client k8s.Interface,
	watchNamespace string,
	isolateNamespace bool,
	podWatch bool,
	resync time.Duration,
) *listers {
	clusterWatch := watchNamespace == api.NamespaceAll
	clusterOption := informers.WithTweakListOptions(nil)
	namespaceOption := informers.WithNamespace(watchNamespace)
	var ingressInformer, resourceInformer, localInformer informers.SharedInformerFactory
	if clusterWatch {
		ingressInformer = informers.NewSharedInformerFactoryWithOptions(client, resync, clusterOption)
		resourceInformer = ingressInformer
	} else if isolateNamespace {
		ingressInformer = informers.NewSharedInformerFactoryWithOptions(client, resync, namespaceOption)
		resourceInformer = ingressInformer
	} else {
		ingressInformer = informers.NewSharedInformerFactoryWithOptions(client, resync, namespaceOption)
		resourceInformer = informers.NewSharedInformerFactoryWithOptions(client, resync, clusterOption)
	}
	if !podWatch || !clusterWatch {
		localInformer = informers.NewSharedInformerFactory(fake.NewSimpleClientset(), 0)
	}
	l := &listers{
		events:   events,
		recorder: recorder,
		logger:   logger,
	}
	l.createIngressLister(ingressInformer.Networking().V1beta1().Ingresses())
	l.createIngressClassLister(ingressInformer.Networking().V1beta1().IngressClasses())
	l.createEndpointLister(resourceInformer.Core().V1().Endpoints())
	l.createServiceLister(resourceInformer.Core().V1().Services())
	l.createSecretLister(resourceInformer.Core().V1().Secrets())
	l.createConfigMapLister(resourceInformer.Core().V1().ConfigMaps())
	if podWatch {
		l.createPodLister(ingressInformer.Core().V1().Pods())
		l.hasPodLister = true
	} else {
		l.createPodLister(localInformer.Core().V1().Pods())
	}
	if clusterWatch {
		// ignoring --disable-node-list
		l.createNodeLister(resourceInformer.Core().V1().Nodes())
		l.hasNodeLister = true
	} else {
		l.createNodeLister(localInformer.Core().V1().Nodes())
	}
	return l
}

func (l *listers) RunAsync(stopCh <-chan struct{}) {
	syncFailed := func() {
		runtime.HandleError(fmt.Errorf("initial cache sync has timed out or shutdown has requested"))
	}
	l.logger.Info("loading object cache...")

	// wait ingress class lister initialize, ingress informers initialization depends on it
	go l.ingressClassInformer.Run(stopCh)
	ingClassSynced := cache.WaitForCacheSync(stopCh,
		l.ingressClassInformer.HasSynced,
	)
	if !ingClassSynced {
		syncFailed()
		return
	}

	// initialize listers and informers
	go l.ingressInformer.Run(stopCh)
	go l.endpointInformer.Run(stopCh)
	go l.serviceInformer.Run(stopCh)
	go l.secretInformer.Run(stopCh)
	go l.configMapInformer.Run(stopCh)
	go l.podInformer.Run(stopCh)
	go l.nodeInformer.Run(stopCh)
	synced := cache.WaitForCacheSync(stopCh,
		l.ingressInformer.HasSynced,
		l.endpointInformer.HasSynced,
		l.serviceInformer.HasSynced,
		l.secretInformer.HasSynced,
		l.configMapInformer.HasSynced,
		l.podInformer.HasSynced,
		l.nodeInformer.HasSynced,
	)
	if synced {
		l.logger.Info("cache successfully synced")
		l.running = true
	} else {
		syncFailed()
	}
}

func (l *listers) createIngressLister(informer informersv1beta1.IngressInformer) {
	l.ingressLister = informer.Lister()
	l.ingressInformer = informer.Informer()
	l.ingressInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			ing := obj.(*networking.Ingress)
			if l.events.IsValidIngress(ing) {
				l.events.Notify(nil, ing)
				if l.running {
					l.recorder.Eventf(ing, api.EventTypeNormal, "CREATE", "Ingress %s/%s", ing.Namespace, ing.Name)
				}
			}
		},
		UpdateFunc: func(old, cur interface{}) {
			if reflect.DeepEqual(old, cur) {
				return
			}
			oldIng := old.(*networking.Ingress)
{'fraction_non_alphanumeric': 0.10326906957250628, 'fraction_numerical': 0.010393964794635373, 'mean_word_length': 4.935820895522388, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7525583', 'n_tokens_mistral': 4279, 'n_tokens_neox': 4073, 'n_words': 748}