Skip to content

Commit 556c75e

Browse files
author
Thibault Gilles
committed
Handle retries and state refresh when dirty
1 parent 9594138 commit 556c75e

File tree

5 files changed

+152
-63
lines changed

5 files changed

+152
-63
lines changed

‎go.mod‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ go 1.12
55
require (
66
github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a // indirect
77
github.com/criteo/haproxy-spoe-go v0.0.0-20190925130734-97891c13d324
8+
github.com/d4l3k/messagediff v1.2.1 // indirect
89
github.com/docker/go-units v0.4.0 // indirect
910
github.com/facebookgo/freeport v0.0.0-20150612182905-d4adf43b75b9
1011
github.com/go-openapi/analysis v0.19.0 // indirect
@@ -25,5 +26,6 @@ require (
2526
golang.org/x/net v0.0.0-20190607181551-461777fb6f67 // indirect
2627
golang.org/x/sys v0.0.0-20190528012530-adf421d2caf4 // indirect
2728
golang.org/x/text v0.3.2 // indirect
29+
gopkg.in/d4l3k/messagediff.v1 v1.2.1
2830
gopkg.in/mcuadros/go-syslog.v2 v2.2.1
2931
)

‎go.sum‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,8 @@ github.com/coredns/coredns v1.1.2 h1:bAFHrSsBeTeRG5W3Nf2su3lUGw7Npw2UKeCJm/3A638
6767
github.com/coredns/coredns v1.1.2/go.mod h1:zASH/MVDgR6XZTbxvOnsZfffS+31vg6Ackf/wo1+AM0=
6868
github.com/criteo/haproxy-spoe-go v0.0.0-20190925130734-97891c13d324 h1:EG4AakHHowlW2TkSX6URMubHsmRjd0HWl4LtI4pD7WA=
6969
github.com/criteo/haproxy-spoe-go v0.0.0-20190925130734-97891c13d324/go.mod h1:3h7I0HgdYy7SIlcSLEUVLpFTfHA0V4qK6QsQEKNLRkI=
70+
github.com/d4l3k/messagediff v1.2.1 h1:ZcAIMYsUg0EAp9X+tt8/enBE/Q8Yd5kzPynLyKptt9U=
71+
github.com/d4l3k/messagediff v1.2.1/go.mod h1:Oozbb1TVXFac9FtSIxHBMnBCq2qeH/2KkEQxENCrlLo=
7072
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
7173
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
7274
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -467,6 +469,8 @@ gopkg.in/asn1-ber.v1 v1.0.0-20181015200546-f715ec2f112d/go.mod h1:cuepJuh7vyXfUy
467469
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
468470
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
469471
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
472+
gopkg.in/d4l3k/messagediff.v1 v1.2.1 h1:70AthpjunwzUiarMHyED52mj9UwtAnE89l1Gmrt3EU0=
473+
gopkg.in/d4l3k/messagediff.v1 v1.2.1/go.mod h1:EUzikiKadqXWcD1AzJLagx0j/BeeWGtn++04Xniyg44=
470474
gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
471475
gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2/go.mod h1:Xk6kEKp8OKb+X14hQBKWaSkCsqBpgog8nAV2xsGOxlo=
472476
gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=

‎haproxy/haproxy.go‎

Lines changed: 15 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -23,14 +23,15 @@ type HAProxy struct {
2323
opts Options
2424
dataplaneClient *dataplane.Dataplane
2525
consulClient *api.Client
26-
cfgC chan consul.Config
27-
currentCfg *consul.Config
2826

29-
oldState state.State
27+
cfgC chan consul.Config
28+
29+
currentConsulConfig *consul.Config
30+
currentHAProxyState state.State
3031

3132
haConfig *haConfig
3233

33-
Ready chan (struct{})
34+
Ready chan struct{}
3435
}
3536

3637
func New(consulClient *api.Client, cfg chan consul.Config, opts Options) *HAProxy {
@@ -49,34 +50,7 @@ func New(consulClient *api.Client, cfg chan consul.Config, opts Options) *HAProx
4950
}
5051

5152
func (h *HAProxy) Run(sd *lib.Shutdown) error {
52-
init := false
53-
statsStarted := false
54-
for {
55-
select {
56-
case c := <-h.cfgC:
57-
if !init {
58-
err := h.start(sd)
59-
if err != nil {
60-
return err
61-
}
62-
init = true
63-
close(h.Ready)
64-
}
65-
err := h.handleChange(c)
66-
if err != nil {
67-
log.Error(err)
68-
}
69-
if !statsStarted {
70-
err = h.startStats()
71-
if err != nil {
72-
log.Error(err)
73-
}
74-
statsStarted = true
75-
}
76-
case <-sd.Stop:
77-
return nil
78-
}
79-
}
53+
return h.watch(sd)
8054
}
8155

8256
func (h *HAProxy) start(sd *lib.Shutdown) error {
@@ -114,6 +88,11 @@ func (h *HAProxy) start(sd *lib.Shutdown) error {
11488
}
11589
h.dataplaneClient = dpc
11690

91+
err = h.startStats()
92+
if err != nil {
93+
log.Error(err)
94+
}
95+
11796
return nil
11897
}
11998

@@ -138,7 +117,7 @@ func (h *HAProxy) startLogger() error {
138117

139118
func (h *HAProxy) startSPOA() error {
140119
spoeAgent := spoe.New(NewSPOEHandler(h.consulClient, func() consul.Config {
141-
return *h.currentCfg
120+
return *h.currentConsulConfig
142121
}).Handler)
143122

144123
lis, err := net.Listen("unix", h.haConfig.SPOESock)
@@ -173,8 +152,8 @@ func (h *HAProxy) startStats() error {
173152

174153
reg := func() {
175154
err = h.consulClient.Agent().ServiceRegister(&api.AgentServiceRegistration{
176-
ID: fmt.Sprintf("%s-connect-stats", h.currentCfg.ServiceID),
177-
Name: fmt.Sprintf("%s-connect-stats", h.currentCfg.ServiceName),
155+
ID: fmt.Sprintf("%s-connect-stats", h.currentConsulConfig.ServiceID),
156+
Name: fmt.Sprintf("%s-connect-stats", h.currentConsulConfig.ServiceName),
178157
Port: port,
179158
Checks: api.AgentServiceChecks{
180159
&api.AgentServiceCheck{
@@ -198,7 +177,7 @@ func (h *HAProxy) startStats() error {
198177
}()
199178
go (&Stats{
200179
dpapi: h.dataplaneClient,
201-
service: h.currentCfg.ServiceName,
180+
service: h.currentConsulConfig.ServiceName,
202181
}).Run()
203182
go func() {
204183
http.Handle("/metrics", promhttp.Handler())

‎haproxy/state.go‎

Lines changed: 122 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1,43 +1,139 @@
11
package haproxy
22

33
import (
4+
"sync/atomic"
5+
"time"
6+
47
"github.com/criteo/haproxy-consul-connect/consul"
58
"github.com/criteo/haproxy-consul-connect/haproxy/state"
9+
"github.com/criteo/haproxy-consul-connect/lib"
610
log "github.com/sirupsen/logrus"
11+
"gopkg.in/d4l3k/messagediff.v1"
712
)
813

9-
func (h *HAProxy) handleChange(cfg consul.Config) error {
10-
log.Info("handling new configuration")
11-
12-
newState, err := state.Generate(state.Options{
13-
EnableIntentions: h.opts.EnableIntentions,
14-
LogRequests: h.opts.LogRequests,
15-
LogSocket: h.haConfig.LogsSock,
16-
SPOEConfigPath: h.haConfig.SPOE,
17-
SPOESocket: h.haConfig.SPOESock,
18-
}, h.haConfig, h.oldState, cfg)
19-
if err != nil {
20-
return err
21-
}
14+
func (h *HAProxy) watch(sd *lib.Shutdown) error {
15+
throttle := time.Tick(50 * time.Millisecond)
16+
currentState := state.State{}
17+
nextState := &atomic.Value{}
18+
next := make(chan struct{}, 1)
19+
dirty := false
2220

23-
tx := h.dataplaneClient.Tnx()
21+
go func() {
22+
for c := range h.cfgC {
23+
select {
24+
case <-sd.Stop:
25+
return
26+
default:
27+
}
2428

25-
log.Debugf("applying new state: %+v", newState)
29+
log.Info("received consul config update")
30+
nextState.Store(c)
31+
h.currentConsulConfig = &c
32+
select {
33+
case next <- struct{}{}:
34+
default:
35+
}
36+
}
37+
}()
2638

27-
err = state.Apply(tx, h.oldState, newState)
28-
if err != nil {
29-
return err
30-
}
39+
go func() {
40+
for range time.Tick(5 * time.Minute) {
41+
select {
42+
case <-sd.Stop:
43+
return
44+
default:
45+
}
46+
47+
dirty = true
48+
}
49+
}()
3150

32-
err = tx.Commit()
33-
if err != nil {
34-
return err
51+
retry := func() {
52+
time.Sleep(3 * time.Second)
53+
select {
54+
case next <- struct{}{}:
55+
default:
56+
}
3557
}
3658

37-
h.oldState = newState
38-
h.currentCfg = &cfg
59+
started := false
60+
for {
61+
for {
62+
select {
63+
case <-sd.Stop:
64+
return nil
65+
case <-next:
66+
}
67+
68+
<-throttle
69+
70+
log.Info("handling new configuration")
71+
if !started {
72+
err := h.start(sd)
73+
if err != nil {
74+
return err
75+
}
76+
started = true
77+
close(h.Ready)
78+
}
79+
80+
if dirty {
81+
log.Info("refreshing haproxy state")
82+
fromHa, err := state.FromHAProxy(h.dataplaneClient)
83+
if err != nil {
84+
log.Errorf("error retrieving haproxy conf: %s", err)
85+
retry()
86+
continue
87+
}
88+
diff, equal := messagediff.PrettyDiff(currentState, fromHa)
89+
if !equal {
90+
log.Errorf("diff found between expected state and haproxy state: %s", diff)
91+
}
92+
currentState = fromHa
93+
dirty = false
94+
}
95+
96+
newConsulCfg := nextState.Load().(consul.Config)
3997

40-
log.Info("state successfully applied")
98+
newState, err := state.Generate(state.Options{
99+
EnableIntentions: h.opts.EnableIntentions,
100+
LogRequests: h.opts.LogRequests,
101+
LogSocket: h.haConfig.LogsSock,
102+
SPOEConfigPath: h.haConfig.SPOE,
103+
SPOESocket: h.haConfig.SPOESock,
104+
}, h.haConfig, currentState, newConsulCfg)
105+
if err != nil {
106+
log.Error(err)
107+
retry()
108+
continue
109+
}
41110

42-
return nil
111+
if currentState.Equal(newState) {
112+
log.Info("no change to apply to haproxy")
113+
continue
114+
}
115+
116+
tx := h.dataplaneClient.Tnx()
117+
118+
log.Debugf("applying new state: %+v", newState)
119+
120+
err = state.Apply(tx, currentState, newState)
121+
if err != nil {
122+
log.Error(err)
123+
retry()
124+
continue
125+
}
126+
127+
err = tx.Commit()
128+
if err != nil {
129+
log.Error(err)
130+
retry()
131+
continue
132+
}
133+
134+
currentState = newState
135+
136+
log.Info("state applied")
137+
}
138+
}
43139
}

‎haproxy/state/state.go‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,10 @@
11
package state
22

3-
import "github.com/haproxytech/models"
3+
import (
4+
"reflect"
5+
6+
"github.com/haproxytech/models"
7+
)
48

59
type FrontendFilter struct {
610
Filter models.Filter
@@ -25,6 +29,10 @@ type State struct {
2529
Backends []Backend
2630
}
2731

32+
func (s State) Equal(o State) bool {
33+
return reflect.DeepEqual(s, o)
34+
}
35+
2836
func (s State) findBackend(name string) (Backend, bool) {
2937
for _, b := range s.Backends {
3038
if b.Backend.Name == name {

0 commit comments

Comments
 (0)