BUG: this commit fixes issues: ip addresses of ingresses updating, creation of ingress delayed

- ip addresses of ingresses updated too slowly with publish service ip address.
- creation of ingress delayed by unnecessary repeating operations or synchronous operations on previous point.

Author: ivanmatmati

main.go

@@ -98,26 +98,24 @@ func main() {
 	publishService := getNamespaceValue(osArgs.PublishService)
 
 	s := store.NewK8sStore(osArgs)
+	k := k8s.New(
+		osArgs,
+		s.NamespacesAccess.Whitelist,
+		publishService,
+	)
 
 	c := controller.NewBuilder().
 		WithHaproxyCfgFile(haproxyConf).
 		WithEventChan(eventChan).
 		WithIngressChan(ingressChan).
 		WithStore(s).
 		WithPublishService(publishService).
+		WithUpdatePublishServiceFunc(k.UpdatePublishService).
+		WithClientSet(k.GetClientset()).
 		WithArgs(osArgs).Build()
 
-	k := k8s.New(
-		osArgs,
-		s.NamespacesAccess.Whitelist,
-		publishService,
-	)
-
 	go k.MonitorChanges(eventChan, ingressChan, stop)
 	go c.Start()
-	if publishService != nil {
-		go ingress.UpdateStatus(k.GetClientset(), s, osArgs.IngressClass, osArgs.EmptyIngressClass, ingressChan, annotations.New())
-	}
 
 	// Catch QUIT signals
 	signalC := make(chan os.Signal, 1)

pkg/controller/builder.go

@@ -10,6 +10,7 @@ import (
 	"github.com/valyala/fasthttp"
 	"github.com/valyala/fasthttp/fasthttpadaptor"
 	"github.com/valyala/fasthttp/pprofhandler"
+	"k8s.io/client-go/kubernetes"
 
 	"github.com/haproxytech/kubernetes-ingress/pkg/annotations"
 	"github.com/haproxytech/kubernetes-ingress/pkg/haproxy"
@@ -24,17 +25,19 @@ import (
 )
 
 type Builder struct {
-	osArgs         utils.OSArgs
-	haproxyClient  api.HAProxyClient
-	haproxyEnv     env.Env
-	haproxyProcess process.Process
-	haproxyRules   rules.Rules
-	haproxyCfgFile []byte
-	annotations    annotations.Annotations
-	store          store.K8s
-	publishService *utils.NamespaceValue
-	eventChan      chan k8s.SyncDataEvent
-	ingressChan    chan ingress.Sync
+	osArgs                   utils.OSArgs
+	haproxyClient            api.HAProxyClient
+	haproxyEnv               env.Env
+	haproxyProcess           process.Process
+	haproxyRules             rules.Rules
+	haproxyCfgFile           []byte
+	annotations              annotations.Annotations
+	store                    store.K8s
+	publishService           *utils.NamespaceValue
+	eventChan                chan k8s.SyncDataEvent
+	ingressChan              chan ingress.Sync
+	updatePublishServiceFunc func(ingresses []*ingress.Ingress, publishServiceAddresses []string)
+	clientSet                *kubernetes.Clientset
 }
 
 var defaultEnv = env.Env{
@@ -114,6 +117,16 @@ func (builder *Builder) WithPublishService(publishService *utils.NamespaceValue)
 	return builder
 }
 
+func (builder *Builder) WithUpdatePublishServiceFunc(updatePublishServiceFunc func(ingresses []*ingress.Ingress, publishServiceAddresses []string)) *Builder {
+	builder.updatePublishServiceFunc = updatePublishServiceFunc
+	return builder
+}
+
+func (builder *Builder) WithClientSet(clientSet *kubernetes.Clientset) *Builder {
+	builder.clientSet = clientSet
+	return builder
+}
+
 func (builder *Builder) Build() *HAProxyController {
 	if builder.haproxyCfgFile == nil {
 		logger.Panic(errors.New("no HAProxy Config file provided"))
@@ -134,17 +147,20 @@ func (builder *Builder) Build() *HAProxyController {
 
 	prefix, errPrefix := utils.GetPodPrefix(os.Getenv("POD_NAME"))
 	logger.Error(errPrefix)
+
+	builder.store.UpdateStatusFunc = ingress.NewStatusIngressUpdater(builder.clientSet, builder.store, builder.osArgs.IngressClass, builder.osArgs.EmptyIngressClass, builder.annotations)
 	return &HAProxyController{
-		osArgs:         builder.osArgs,
-		haproxy:        haproxy,
-		podNamespace:   os.Getenv("POD_NAMESPACE"),
-		podPrefix:      prefix,
-		store:          builder.store,
-		eventChan:      builder.eventChan,
-		ingressChan:    builder.ingressChan,
-		publishService: builder.publishService,
-		annotations:    builder.annotations,
-		chShutdown:     chShutdown,
+		osArgs:                   builder.osArgs,
+		haproxy:                  haproxy,
+		podNamespace:             os.Getenv("POD_NAMESPACE"),
+		podPrefix:                prefix,
+		store:                    builder.store,
+		eventChan:                builder.eventChan,
+		ingressChan:              builder.ingressChan,
+		publishService:           builder.publishService,
+		annotations:              builder.annotations,
+		chShutdown:               chShutdown,
+		updatePublishServiceFunc: builder.updatePublishServiceFunc,
 	}
 }
 

pkg/controller/controller.go

@@ -34,21 +34,22 @@ var logger = utils.GetLogger()
 
 // HAProxyController is ingress controller
 type HAProxyController struct {
-	haproxy        haproxy.HAProxy
-	osArgs         utils.OSArgs
-	store          store.K8s
-	annotations    annotations.Annotations
-	publishService *utils.NamespaceValue
-	auxCfgModTime  int64
-	eventChan      chan k8s.SyncDataEvent
-	ingressChan    chan ingress.Sync
-	ready          bool
-	reload         bool
-	restart        bool
-	updateHandlers []UpdateHandler
-	podNamespace   string
-	podPrefix      string
-	chShutdown     chan struct{}
+	haproxy                  haproxy.HAProxy
+	osArgs                   utils.OSArgs
+	store                    store.K8s
+	annotations              annotations.Annotations
+	publishService           *utils.NamespaceValue
+	auxCfgModTime            int64
+	eventChan                chan k8s.SyncDataEvent
+	ingressChan              chan ingress.Sync
+	ready                    bool
+	reload                   bool
+	restart                  bool
+	updateHandlers           []UpdateHandler
+	podNamespace             string
+	podPrefix                string
+	chShutdown               chan struct{}
+	updatePublishServiceFunc func(ingresses []*ingress.Ingress, publishServiceAddresses []string)
 }
 
 // Wrapping a Native-Client transaction and commit it.
@@ -110,27 +111,29 @@ func (c *HAProxyController) updateHAProxy() {
 		logger.Error(route.CustomRoutesReset(c.haproxy))
 	}
 
+	ingresses := []*ingress.Ingress{}
 	for _, namespace := range c.store.Namespaces {
 		if !namespace.Relevant {
 			continue
 		}
+		c.store.SecretsProcessed = map[string]struct{}{}
 		for _, ingResource := range namespace.Ingresses {
 			i := ingress.New(c.store, ingResource, c.osArgs.IngressClass, c.osArgs.EmptyIngressClass, c.annotations)
 			if i == nil {
 				logger.Debugf("ingress '%s/%s' ignored: no matching IngressClass", ingResource.Namespace, ingResource.Name)
 				continue
 			}
-			if c.publishService != nil && ingResource.Status == store.ADDED {
-				select {
-				case c.ingressChan <- ingress.Sync{Ingress: ingResource}:
-				default:
-					logger.Errorf("Ingress %s/%s: unable to sync status: sync channel full", ingResource.Namespace, ingResource.Name)
-				}
+			if ingResource.Status == store.ADDED {
+				ingresses = append(ingresses, i)
 			}
 			c.reload = i.Update(c.store, c.haproxy, c.annotations) || c.reload
 		}
 	}
 
+	if len(ingresses) > 0 {
+		go c.updatePublishServiceFunc(ingresses, c.store.PublishServiceAddresses)
+	}
+
 	for _, handler := range c.updateHandlers {
 		reload, err = handler.Update(c.store, c.haproxy, c.annotations)
 		logger.Error(err)

pkg/controller/monitor.go

@@ -71,6 +71,8 @@ func (c *HAProxyController) SyncData() {
 			change = c.store.EventSecret(ns, job.Data.(*store.Secret))
 		case k8s.POD:
 			change = c.store.EventPod(job.Data.(store.PodEvent))
+		case k8s.PUBLISH_SERVICE:
+			change = c.store.EventPublishService(ns, job.Data.(*store.Service))
 		}
 		hadChanges = hadChanges || change
 	}

pkg/ingress/ingress.go

@@ -129,7 +129,12 @@ func (i *Ingress) handlePath(k store.K8s, h haproxy.HAProxy, host string, path *
 		return
 	}
 	// Endpoints
-	endpointsReload := svc.HandleHAProxySrvs(k, h)
+	service := svc.GetResource()
+	var endpointsReload bool
+	if _, ok := k.ServiceProcessed[service.Namespace+"/"+service.Name]; !ok {
+		endpointsReload = svc.HandleHAProxySrvs(k, h)
+		k.ServiceProcessed[service.Namespace+"/"+service.Name] = struct{}{}
+	}
 	return backendReload || endpointsReload || routeReload, err
 }
 
@@ -207,11 +212,15 @@ func (i *Ingress) Update(k store.K8s, h haproxy.HAProxy, a annotations.Annotatio
 	// Ingress secrets
 	logger.Tracef("Ingress '%s/%s': processing secrets...", i.resource.Namespace, i.resource.Name)
 	for _, tls := range i.resource.TLS {
+		if _, ok := k.SecretsProcessed[i.resource.Namespace+"/"+tls.SecretName]; ok {
+			continue
+		}
 		secret, secErr := k.GetSecret(i.resource.Namespace, tls.SecretName)
 		if secErr != nil {
 			logger.Warningf("Ingress '%s/%s': %s", i.resource.Namespace, i.resource.Name, secErr)
 			continue
 		}
+		k.SecretsProcessed[i.resource.Namespace+"/"+tls.SecretName] = struct{}{}
 		_, err := h.AddSecret(secret, certs.FT_CERT)
 		logger.Error(err)
 	}

pkg/ingress/status.go

@@ -5,89 +5,37 @@ import (
 	"fmt"
 	"net"
 
+	"github.com/haproxytech/kubernetes-ingress/pkg/annotations"
+	"github.com/haproxytech/kubernetes-ingress/pkg/store"
+	"github.com/haproxytech/kubernetes-ingress/pkg/utils"
 	corev1 "k8s.io/api/core/v1"
 	extensionsv1beta1 "k8s.io/api/extensions/v1beta1"
 	networkingv1 "k8s.io/api/networking/v1"
 	networkingv1beta "k8s.io/api/networking/v1beta1"
 	k8serror "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
-
-	"github.com/haproxytech/kubernetes-ingress/pkg/annotations"
-	"github.com/haproxytech/kubernetes-ingress/pkg/store"
 )
 
-func UpdateStatus(client *kubernetes.Clientset, k store.K8s, class string, emptyClass bool, channel chan Sync, a annotations.Annotations) {
-	var i *Ingress
-	addresses := []string{}
-	for sync := range channel {
-		// Published Service updated: Update all Ingresses
-		if sync.Service != nil && getServiceAddresses(sync.Service, &addresses) {
-			logger.Debug("Addresses of Ingress Controller service changed, status of all ingress resources are going to be updated")
-			for _, ns := range k.Namespaces {
-				if !ns.Relevant {
-					continue
-				}
-				for _, ingress := range k.Namespaces[ns.Name].Ingresses {
-					if i = New(k, ingress, class, emptyClass, a); i != nil {
-						logger.Error(i.updateStatus(client, addresses))
-					}
-				}
+type UpdateStatus func(ingresses []*store.Ingress, publishServiceAddresses []string)
+
+func NewStatusIngressUpdater(client *kubernetes.Clientset, k store.K8s, class string, emptyClass bool, a annotations.Annotations) UpdateStatus {
+	return func(ingresses []*store.Ingress, publishServiceAddresses []string) {
+		for _, ingress := range ingresses {
+			if ing := New(k, ingress, class, emptyClass, a); ing != nil {
+				logger.Error(ing.UpdateStatus(client, publishServiceAddresses))
 			}
-		} else if i = New(k, sync.Ingress, class, emptyClass, a); i != nil {
-			// Update single Ingress
-			logger.Error(i.updateStatus(client, addresses))
 		}
 	}
 }
 
-func getServiceAddresses(service *corev1.Service, curAddr *[]string) (updated bool) {
-	addresses := []string{}
-	switch service.Spec.Type {
-	case corev1.ServiceTypeExternalName:
-		addresses = []string{service.Spec.ExternalName}
-	case corev1.ServiceTypeClusterIP:
-		addresses = []string{service.Spec.ClusterIP}
-	case corev1.ServiceTypeNodePort:
-		if service.Spec.ExternalIPs != nil {
-			addresses = append(addresses, service.Spec.ExternalIPs...)
-		} else {
-			addresses = append(addresses, service.Spec.ClusterIP)
-		}
-	case corev1.ServiceTypeLoadBalancer:
-		for _, ip := range service.Status.LoadBalancer.Ingress {
-			if ip.IP == "" {
-				addresses = append(addresses, ip.Hostname)
-			} else {
-				addresses = append(addresses, ip.IP)
-			}
-		}
-		addresses = append(addresses, service.Spec.ExternalIPs...)
-	default:
-		logger.Errorf("Unable to extract IP address/es from service %s/%s", service.Namespace, service.Name)
-		return
-	}
+func (i *Ingress) UpdateStatus(client *kubernetes.Clientset, addresses []string) (err error) {
+	var lbi []corev1.LoadBalancerIngress
 
-	if len(*curAddr) != len(addresses) {
-		updated = true
-		*curAddr = addresses
+	if utils.EqualSliceStringsWithoutOrder(i.resource.Addresses, addresses) {
 		return
 	}
-	for i, address := range addresses {
-		if address != (*curAddr)[i] {
-			updated = true
-			break
-		}
-	}
-	if updated {
-		*curAddr = addresses
-	}
-	return
-}
 
-func (i *Ingress) updateStatus(client *kubernetes.Clientset, addresses []string) (err error) {
-	logger.Tracef("Updating status of Ingress %s/%s", i.resource.Namespace, i.resource.Name)
-	var lbi []corev1.LoadBalancerIngress
 	for _, addr := range addresses {
 		if net.ParseIP(addr) == nil {
 			lbi = append(lbi, corev1.LoadBalancerIngress{Hostname: addr})
@@ -135,6 +83,13 @@ func (i *Ingress) updateStatus(client *kubernetes.Clientset, addresses []string)
 		return fmt.Errorf("failed to update LoadBalancer status of ingress %s/%s: %w", i.resource.Namespace, i.resource.Name, err)
 	}
 	logger.Tracef("Successful update of LoadBalancer status in ingress %s/%s", i.resource.Namespace, i.resource.Name)
-
+	// Allow to store the publish service addresses affected to the ingress for future comparison in update test.
+	i.resource.Addresses = addresses
 	return nil
 }
+
+func UpdatePublishService(ingresses []*Ingress, api *kubernetes.Clientset, publishServiceAddresses []string) {
+	for _, i := range ingresses {
+		logger.Error(i.UpdateStatus(api, publishServiceAddresses))
+	}
+}