Skip to content

Commit ea7b243

Browse files
ivanmatmatiivanmatmati
committed
BUG: this commit fixes the two following related issues:
- 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.
1 parent 2865276 commit ea7b243

File tree

13 files changed

+414
-272
lines changed

13 files changed

+414
-272
lines changed

‎controller/controller.go

+10-11
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,8 @@ func (c *HAProxyController) Start() {
116116
logger.Printf("Running on Kubernetes version: %s %s", k8sVersion.String(), k8sVersion.Platform)
117117
}
118118

119+
c.Store.UpdateStatusFunc = ingress.NewStatusIngressUpdater(c.k8s.API, c.Store, c.OSArgs.IngressClass, c.OSArgs.EmptyIngressClass)
120+
119121
// Monitor k8s events
120122
var chanSize int64 = int64(watch.DefaultChanSize * 6)
121123
if c.OSArgs.ChannelSize > 0 {
@@ -124,11 +126,6 @@ func (c *HAProxyController) Start() {
124126
logger.Infof("Channel size: %d", chanSize)
125127
c.eventChan = make(chan SyncDataEvent, chanSize)
126128
go c.monitorChanges()
127-
if c.PublishService != nil {
128-
// Update Ingress status
129-
c.ingressChan = make(chan ingress.Sync, chanSize)
130-
go ingress.UpdateStatus(c.k8s.API, c.Store, c.OSArgs.IngressClass, c.OSArgs.EmptyIngressClass, c.ingressChan)
131-
}
132129
}
133130

134131
// Stop handles shutting down HAProxyController
@@ -141,6 +138,7 @@ func (c *HAProxyController) Stop() {
141138
func (c *HAProxyController) updateHAProxy() {
142139
var reload bool
143140
var err error
141+
144142
logger.Trace("HAProxy config sync started")
145143

146144
err = c.Client.APIStartTransaction()
@@ -160,26 +158,27 @@ func (c *HAProxyController) updateHAProxy() {
160158
logger.Error(route.CustomRoutesReset(c.Client))
161159
}
162160

161+
ingresses := []*ingress.Ingress{}
163162
for _, namespace := range c.Store.Namespaces {
164163
if !namespace.Relevant {
165164
continue
166165
}
166+
c.Store.SecretsProcessed = map[string]struct{}{}
167167
for _, ingResource := range namespace.Ingresses {
168168
i := ingress.New(c.Store, ingResource, c.OSArgs.IngressClass, c.OSArgs.EmptyIngressClass)
169169
if i == nil {
170170
logger.Debugf("ingress '%s/%s' ignored: no matching IngressClass", ingResource.Namespace, ingResource.Name)
171171
continue
172172
}
173-
if c.PublishService != nil && ingResource.Status == ADDED {
174-
select {
175-
case c.ingressChan <- ingress.Sync{Ingress: ingResource}:
176-
default:
177-
logger.Errorf("Ingress %s/%s: unable to sync status: sync channel full", ingResource.Namespace, ingResource.Name)
178-
}
173+
if ingResource.Status == ADDED {
174+
ingresses = append(ingresses, i)
179175
}
180176
c.reload = i.Update(c.Store, &c.Cfg, c.Client) || c.reload
181177
}
182178
}
179+
if len(ingresses) > 0 {
180+
go ingress.UpdatePublishService(ingresses, c.k8s.API, c.Store.PublishServiceAddresses)
181+
}
183182

184183
for _, handler := range c.updateHandlers {
185184
reload, err = handler.Update(c.Store, &c.Cfg, c.Client)

‎controller/ingress/ingress.go

+16-1
Original file line numberDiff line numberDiff line change
@@ -87,12 +87,15 @@ func (i *Ingress) handlePath(k store.K8s, cfg *configuration.ControllerCfg, api
8787
if err != nil {
8888
return
8989
}
90+
9091
// Backend
9192
backendReload, err := svc.HandleBackend(api, k)
9293
if err != nil {
9394
return
9495
}
96+
9597
backendName, _ := svc.GetBackendName()
98+
9699
// Route
97100
var routeReload bool
98101
ingRoute := route.Route{
@@ -119,7 +122,12 @@ func (i *Ingress) handlePath(k store.K8s, cfg *configuration.ControllerCfg, api
119122
}
120123
cfg.ActiveBackends[backendName] = struct{}{}
121124
// Endpoints
122-
endpointsReload := svc.HandleHAProxySrvs(api, k)
125+
service := svc.GetResource()
126+
var endpointsReload bool
127+
if _, ok := k.ServiceProcessed[service.Namespace+"/"+service.Name]; !ok {
128+
endpointsReload = svc.HandleHAProxySrvs(api, k)
129+
k.ServiceProcessed[service.Namespace+"/"+service.Name] = struct{}{}
130+
}
123131
return backendReload || endpointsReload || routeReload, err
124132
}
125133

@@ -192,14 +200,19 @@ func (i *Ingress) Update(k store.K8s, cfg *configuration.ControllerCfg, api api.
192200
logger.Infof("Setting http default backend to '%s'", backendName)
193201
}
194202
}
203+
195204
// Ingress secrets
196205
logger.Tracef("Ingress '%s/%s': processing secrets...", i.resource.Namespace, i.resource.Name)
197206
for _, tls := range i.resource.TLS {
207+
if _, ok := k.SecretsProcessed[i.resource.Namespace+"/"+tls.SecretName]; ok {
208+
continue
209+
}
198210
secret, secErr := k.GetSecret(i.resource.Namespace, tls.SecretName)
199211
if secErr != nil {
200212
logger.Warningf("Ingress '%s/%s': %s", i.resource.Namespace, i.resource.Name, secErr)
201213
continue
202214
}
215+
k.SecretsProcessed[i.resource.Namespace+"/"+tls.SecretName] = struct{}{}
203216
_, err := cfg.Certificates.HandleTLSSecret(secret, certs.FT_CERT)
204217
logger.Error(err)
205218
}
@@ -217,6 +230,7 @@ func (i *Ingress) Update(k store.K8s, cfg *configuration.ControllerCfg, api api.
217230
cfg.SSLPassthrough = true
218231
}
219232
i.HandleAnnotations(k, cfg)
233+
220234
// Ingress rules
221235
logger.Tracef("ingress '%s/%s': processing rules...", i.resource.Namespace, i.resource.Name)
222236
for _, rule := range i.resource.Rules {
@@ -228,5 +242,6 @@ func (i *Ingress) Update(k store.K8s, cfg *configuration.ControllerCfg, api api.
228242
}
229243
}
230244
}
245+
231246
return
232247
}

‎controller/ingress/status.go

+19-63
Original file line numberDiff line numberDiff line change
@@ -14,79 +14,28 @@ import (
1414
"k8s.io/client-go/kubernetes"
1515

1616
"github.com/haproxytech/kubernetes-ingress/controller/store"
17+
"github.com/haproxytech/kubernetes-ingress/controller/utils"
1718
)
1819

19-
func UpdateStatus(client *kubernetes.Clientset, k store.K8s, class string, emptyClass bool, channel chan Sync) {
20-
var i *Ingress
21-
addresses := []string{}
22-
for sync := range channel {
23-
// Published Service updated: Update all Ingresses
24-
if sync.Service != nil && getServiceAddresses(sync.Service, &addresses) {
25-
logger.Debug("Addresses of Ingress Controller service changed, status of all ingress resources are going to be updated")
26-
for _, ns := range k.Namespaces {
27-
if !ns.Relevant {
28-
continue
29-
}
30-
for _, ingress := range k.Namespaces[ns.Name].Ingresses {
31-
if i = New(k, ingress, class, emptyClass); i != nil {
32-
logger.Error(i.updateStatus(client, addresses))
33-
}
34-
}
20+
type UpdateStatus func(service store.Service, ingresses []*store.Ingress)
21+
22+
func NewStatusIngressUpdater(client *kubernetes.Clientset, k store.K8s, class string, emptyClass bool) UpdateStatus {
23+
return func(service store.Service, ingresses []*store.Ingress) {
24+
for _, ingress := range ingresses {
25+
if ing := New(k, ingress, class, emptyClass); ing != nil {
26+
logger.Error(ing.UpdateStatus(client, service.Addresses))
3527
}
36-
} else if i = New(k, sync.Ingress, class, emptyClass); i != nil {
37-
// Update single Ingress
38-
logger.Error(i.updateStatus(client, addresses))
3928
}
4029
}
4130
}
4231

43-
func getServiceAddresses(service *corev1.Service, curAddr *[]string) (updated bool) {
44-
addresses := []string{}
45-
switch service.Spec.Type {
46-
case corev1.ServiceTypeExternalName:
47-
addresses = []string{service.Spec.ExternalName}
48-
case corev1.ServiceTypeClusterIP:
49-
addresses = []string{service.Spec.ClusterIP}
50-
case corev1.ServiceTypeNodePort:
51-
if service.Spec.ExternalIPs != nil {
52-
addresses = append(addresses, service.Spec.ExternalIPs...)
53-
} else {
54-
addresses = append(addresses, service.Spec.ClusterIP)
55-
}
56-
case corev1.ServiceTypeLoadBalancer:
57-
for _, ip := range service.Status.LoadBalancer.Ingress {
58-
if ip.IP == "" {
59-
addresses = append(addresses, ip.Hostname)
60-
} else {
61-
addresses = append(addresses, ip.IP)
62-
}
63-
}
64-
addresses = append(addresses, service.Spec.ExternalIPs...)
65-
default:
66-
logger.Errorf("Unable to extract IP address/es from service %s/%s", service.Namespace, service.Name)
67-
return
68-
}
32+
func (i *Ingress) UpdateStatus(client *kubernetes.Clientset, addresses []string) (err error) {
33+
var lbi []corev1.LoadBalancerIngress
6934

70-
if len(*curAddr) != len(addresses) {
71-
updated = true
72-
*curAddr = addresses
35+
if utils.EqualSliceStringsWithoutOrder(i.resource.Addresses, addresses) {
7336
return
7437
}
75-
for i, address := range addresses {
76-
if address != (*curAddr)[i] {
77-
updated = true
78-
break
79-
}
80-
}
81-
if updated {
82-
*curAddr = addresses
83-
}
84-
return
85-
}
8638

87-
func (i *Ingress) updateStatus(client *kubernetes.Clientset, addresses []string) (err error) {
88-
logger.Tracef("Updating status of Ingress %s/%s", i.resource.Namespace, i.resource.Name)
89-
var lbi []corev1.LoadBalancerIngress
9039
for _, addr := range addresses {
9140
if net.ParseIP(addr) == nil {
9241
lbi = append(lbi, corev1.LoadBalancerIngress{Hostname: addr})
@@ -134,6 +83,13 @@ func (i *Ingress) updateStatus(client *kubernetes.Clientset, addresses []string)
13483
return fmt.Errorf("failed to update LoadBalancer status of ingress %s/%s: %w", i.resource.Namespace, i.resource.Name, err)
13584
}
13685
logger.Tracef("Successful update of LoadBalancer status in ingress %s/%s", i.resource.Namespace, i.resource.Name)
137-
86+
// Allow to store the publish service addresses affected to the ingress for future comparison in update test.
87+
i.resource.Addresses = addresses
13888
return nil
13989
}
90+
91+
func UpdatePublishService(ingresses []*Ingress, api *kubernetes.Clientset, publishServiceAddresses []string) {
92+
for _, i := range ingresses {
93+
logger.Error(i.UpdateStatus(api, publishServiceAddresses))
94+
}
95+
}

‎controller/kubernetes.go

+49-9
Original file line numberDiff line numberDiff line change
@@ -474,8 +474,11 @@ func (k *K8s) EventsServices(channel chan SyncDataEvent, ingressChan chan ingres
474474
}
475475
k.Logger.Tracef("%s %s: %s", SERVICE, item.Status, item.Name)
476476
channel <- SyncDataEvent{SyncType: SERVICE, Namespace: item.Namespace, Data: item}
477-
if publishSvc != nil && publishSvc.Namespace == data.Namespace && publishSvc.Name == data.Name {
478-
ingressChan <- ingress.Sync{Service: data}
477+
if publishSvc != nil && publishSvc.Namespace == item.Namespace && publishSvc.Name == item.Name {
478+
// item copy because of ADDED handler in events.go which must modify the STATUS based solely on addresses
479+
itemCopy := *item
480+
itemCopy.Addresses = getServiceAddresses(data)
481+
channel <- SyncDataEvent{SyncType: PUBLISH_SERVICE, Namespace: item.Namespace, Data: &itemCopy}
479482
}
480483
},
481484
DeleteFunc: func(obj interface{}) {
@@ -499,8 +502,9 @@ func (k *K8s) EventsServices(channel chan SyncDataEvent, ingressChan chan ingres
499502
}
500503
k.Logger.Tracef("%s %s: %s", SERVICE, item.Status, item.Name)
501504
channel <- SyncDataEvent{SyncType: SERVICE, Namespace: item.Namespace, Data: item}
502-
if publishSvc != nil && publishSvc.Namespace == data.Namespace && publishSvc.Name == data.Name {
503-
ingressChan <- ingress.Sync{Service: data}
505+
if publishSvc != nil && publishSvc.Namespace == item.Namespace && publishSvc.Name == item.Name {
506+
item.Addresses = getServiceAddresses(data)
507+
channel <- SyncDataEvent{SyncType: PUBLISH_SERVICE, Namespace: data.Namespace, Data: item}
504508
}
505509
},
506510
UpdateFunc: func(oldObj, newObj interface{}) {
@@ -522,9 +526,7 @@ func (k *K8s) EventsServices(channel chan SyncDataEvent, ingressChan chan ingres
522526
k.Logger.Tracef("forwarding to ExternalName Services for %v is disabled", data2)
523527
return
524528
}
525-
if publishSvc != nil && publishSvc.Namespace == data2.Namespace && publishSvc.Name == data2.Name {
526-
ingressChan <- ingress.Sync{Service: data2}
527-
}
529+
528530
status := MODIFIED
529531
item1 := &store.Service{
530532
Namespace: data1.GetNamespace(),
@@ -564,8 +566,15 @@ func (k *K8s) EventsServices(channel chan SyncDataEvent, ingressChan chan ingres
564566
if item2.Equal(item1) {
565567
return
566568
}
567-
k.Logger.Tracef("%s %s: %s", SERVICE, item2.Status, item2.Name)
568-
channel <- SyncDataEvent{SyncType: SERVICE, Namespace: item2.Namespace, Data: item2}
569+
570+
if !item2.Equal(item1) {
571+
k.Logger.Tracef("%s %s: %s", SERVICE, item2.Status, item2.Name)
572+
channel <- SyncDataEvent{SyncType: SERVICE, Namespace: item2.Namespace, Data: item2}
573+
}
574+
if publishSvc != nil && publishSvc.Namespace == item2.Namespace && publishSvc.Name == item2.Name {
575+
item2.Addresses = getServiceAddresses(data2)
576+
channel <- SyncDataEvent{SyncType: PUBLISH_SERVICE, Namespace: item2.Namespace, Data: item2}
577+
}
569578
},
570579
})
571580
go informer.Run(stop)
@@ -766,3 +775,34 @@ func (k *K8s) IsNetworkingV1ApiSupported() bool {
766775

767776
return major == 1 && minor >= 19
768777
}
778+
779+
func getServiceAddresses(service *corev1.Service) (addresses []string) {
780+
switch service.Spec.Type {
781+
case corev1.ServiceTypeExternalName:
782+
addresses = []string{service.Spec.ExternalName}
783+
case corev1.ServiceTypeClusterIP:
784+
addresses = []string{service.Spec.ClusterIP}
785+
case corev1.ServiceTypeNodePort:
786+
if service.Spec.ExternalIPs != nil {
787+
addresses = append(addresses, service.Spec.ExternalIPs...)
788+
} else {
789+
addresses = append(addresses, service.Spec.ClusterIP)
790+
}
791+
case corev1.ServiceTypeLoadBalancer:
792+
for _, ip := range service.Status.LoadBalancer.Ingress {
793+
if ip.IP == "" {
794+
addresses = append(addresses, ip.Hostname)
795+
} else {
796+
addresses = append(addresses, ip.IP)
797+
}
798+
}
799+
addresses = append(addresses, service.Spec.ExternalIPs...)
800+
default:
801+
logger.Errorf("Unable to extract IP address/es from service %s/%s", service.Namespace, service.Name)
802+
return
803+
}
804+
if addresses == nil {
805+
addresses = []string{}
806+
}
807+
return
808+
}

‎controller/monitor.go

+2
Original file line numberDiff line numberDiff line change
@@ -126,6 +126,8 @@ func (c *HAProxyController) SyncData() {
126126
change = c.Store.EventSecret(ns, job.Data.(*store.Secret))
127127
case POD:
128128
change = c.Store.EventPod(job.Data.(store.PodEvent))
129+
case PUBLISH_SERVICE:
130+
change = c.Store.EventPublishService(ns, job.Data.(*store.Service))
129131
}
130132
hadChanges = hadChanges || change
131133
}

‎controller/service/endpoints.go

+3
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,9 @@ import (
2828
// HandleHAProxySrvs handles the haproxy backend servers of the corresponding IngressPath (service + port)
2929
func (s *Service) HandleHAProxySrvs(client api.HAProxyClient, store store.K8s) (reload bool) {
3030
var srvsScaled bool
31+
3132
backend, err := s.getRuntimeBackend(store)
33+
3234
if err != nil {
3335
logger.Warningf("Ingress '%s/%s': %s", s.resource.Namespace, s.resource.Name, err)
3436
if servers, _ := client.BackendServersGet(s.backend.Name); servers != nil {
@@ -47,6 +49,7 @@ func (s *Service) HandleHAProxySrvs(client api.HAProxyClient, store store.K8s) (
4749
s.updateHAProxySrv(client, *srvSlot, backend.Endpoints.Port)
4850
}
4951
}
52+
5053
if backend.DynUpdateFailed {
5154
backend.DynUpdateFailed = false
5255
return true

0 commit comments

Comments
 (0)