1
=== added directory 'worker/firewaller'
2
=== added file 'worker/firewaller/export_test.go'
3
--- worker/firewaller/export_test.go	1970-01-01 00:00:00 +0000
4
+++ worker/firewaller/export_test.go	2012-07-18 16:28:20 +0000
5
@@ -0,0 +1,16 @@
6
1
package firewaller
7
2
8
3
// AllMachines returns the ids of all monitored machines.
9
4
func (fw *Firewaller) AllMachines() []int {
10
5
	all := []int{}
11
6
	for _, machine := range fw.machines {
12
7
		all = append(all, machine.id)
13
8
	}
14
9
	return all
15
10
}
16
11
17
12
// CloseState allows to close the state of the firewaller
18
13
// externally.
19
14
func (fw *Firewaller) CloseState() error {
20
15
	return fw.st.Close()
21
16
}
22
0
17
23
=== added file 'worker/firewaller/firewaller.go'
24
--- worker/firewaller/firewaller.go	1970-01-01 00:00:00 +0000
25
+++ worker/firewaller/firewaller.go	2012-07-18 16:28:20 +0000
26
@@ -0,0 +1,154 @@
27
1
package firewaller
28
2
29
3
import (
30
4
	"launchpad.net/juju-core/log"
31
5
	"launchpad.net/juju-core/state"
32
6
	"launchpad.net/juju-core/state/watcher"
33
7
	"launchpad.net/tomb"
34
8
)
35
9
36
10
// Firewaller watches the state for ports opened or closed
37
11
// and reflects those changes onto the backing environment.
38
12
type Firewaller struct {
39
13
	st                  *state.State
40
14
	tomb                tomb.Tomb
41
15
	machinesWatcher     *state.MachinesWatcher
42
16
	machines            map[int]*machineTracker
43
17
	machineUnitsChanges chan *machineUnitsChange
44
18
}
45
19
46
20
// NewFirewaller returns a new Firewaller.
47
21
func NewFirewaller(st *state.State) (*Firewaller, error) {
48
22
	fw := &Firewaller{
49
23
		st:                  st,
50
24
		machinesWatcher:     st.WatchMachines(),
51
25
		machines:            make(map[int]*machineTracker),
52
26
		machineUnitsChanges: make(chan *machineUnitsChange),
53
27
	}
54
28
	go fw.loop()
55
29
	return fw, nil
56
30
}
57
31
58
32
func (fw *Firewaller) loop() {
59
33
	defer fw.finish()
60
34
	for {
61
35
		select {
62
36
		case <-fw.tomb.Dying():
63
37
			return
64
38
		case change, ok := <-fw.machinesWatcher.Changes():
65
39
			if !ok {
66
40
				return
67
41
			}
68
42
			for _, removedMachine := range change.Removed {
69
43
				m, ok := fw.machines[removedMachine.Id()]
70
44
				if !ok {
71
45
					panic("trying to remove machine that wasn't added")
72
46
				}
73
47
				delete(fw.machines, removedMachine.Id())
74
48
				if err := m.stop(); err != nil {
75
49
					log.Printf("machine tracker %d returned error when stopping: %v", removedMachine.Id(), err)
76
50
				}
77
51
				log.Debugf("firewaller: remove-machine %v", removedMachine.Id())
78
52
			}
79
53
			for _, addedMachine := range change.Added {
80
54
				m := newMachineTracker(addedMachine, fw)
81
55
				fw.machines[addedMachine.Id()] = m
82
56
				log.Debugf("firewaller: add-machine %v", m.id)
83
57
			}
84
58
		case <-fw.machineUnitsChanges:
85
59
			// TODO(mue) fill with life.
86
60
		}
87
61
	}
88
62
}
89
63
90
64
// finishes cleans up when the firewaller is stopping.
91
65
func (fw *Firewaller) finish() {
92
66
	watcher.Stop(fw.machinesWatcher, &fw.tomb)
93
67
	for _, m := range fw.machines {
94
68
		fw.tomb.Kill(m.stop())
95
69
	}
96
70
	fw.tomb.Done()
97
71
}
98
72
99
73
// Wait waits for the Firewaller to exit.
100
74
func (fw *Firewaller) Wait() error {
101
75
	return fw.tomb.Wait()
102
76
}
103
77
104
78
// Stop stops the Firewaller and returns any error encountered while stopping.
105
79
func (fw *Firewaller) Stop() error {
106
80
	fw.tomb.Kill(nil)
107
81
	return fw.tomb.Wait()
108
82
}
109
83
110
84
// machineUnitsChange contains the changed units for one specific machine. 
111
85
type machineUnitsChange struct {
112
86
	machine *machineTracker
113
87
	change  *state.MachineUnitsChange
114
88
}
115
89
116
90
// machineTracker keeps track of the unit changes of a machine.
117
91
type machineTracker struct {
118
92
	tomb       tomb.Tomb
119
93
	firewaller *Firewaller
120
94
	id         int
121
95
	watcher    *state.MachineUnitsWatcher
122
96
	ports      map[state.Port]*unitTracker
123
97
}
124
98
125
99
// newMachineTracker creates a new machine tracker keeping track of
126
100
// unit changes of the passed machine.
127
101
func newMachineTracker(mst *state.Machine, fw *Firewaller) *machineTracker {
128
102
	mt := &machineTracker{
129
103
		firewaller: fw,
130
104
		id:         mst.Id(),
131
105
		watcher:    mst.WatchUnits(),
132
106
		ports:      make(map[state.Port]*unitTracker),
133
107
	}
134
108
	go mt.loop()
135
109
	return mt
136
110
}
137
111
138
112
// loop is the backend watching for machine units changes.
139
113
func (mt *machineTracker) loop() {
140
114
	defer mt.tomb.Done()
141
115
	defer mt.watcher.Stop()
142
116
	for {
143
117
		select {
144
118
		case <-mt.firewaller.tomb.Dying():
145
119
			return
146
120
		case <-mt.tomb.Dying():
147
121
			return
148
122
		case change, ok := <-mt.watcher.Changes():
149
123
			// Send change or nil.
150
124
			select {
151
125
			case mt.firewaller.machineUnitsChanges <- &machineUnitsChange{mt, change}:
152
126
			case <-mt.firewaller.tomb.Dying():
153
127
				return
154
128
			case <-mt.tomb.Dying():
155
129
				return
156
130
			}
157
131
			// The watcher terminated prematurely, so end the loop.
158
132
			if !ok {
159
133
				mt.firewaller.tomb.Kill(watcher.MustErr(mt.watcher))
160
134
				return
161
135
			}
162
136
		}
163
137
	}
164
138
}
165
139
166
140
// stop stops the machine tracker.
167
141
func (mt *machineTracker) stop() error {
168
142
	mt.tomb.Kill(nil)
169
143
	return mt.tomb.Wait()
170
144
}
171
145
172
146
type unitTracker struct {
173
147
	service *serviceTracker
174
148
	id      string
175
149
	ports   []state.Port
176
150
}
177
151
178
152
type serviceTracker struct {
179
153
	exposed bool
180
154
}
181
0
155
182
=== added file 'worker/firewaller/firewaller_test.go'
183
--- worker/firewaller/firewaller_test.go	1970-01-01 00:00:00 +0000
184
+++ worker/firewaller/firewaller_test.go	2012-07-18 16:28:20 +0000
185
@@ -0,0 +1,140 @@
186
1
package firewaller_test
187
2
188
3
import (
189
4
	"fmt"
190
5
	. "launchpad.net/gocheck"
191
6
	"launchpad.net/juju-core/environs/dummy"
192
7
	"launchpad.net/juju-core/log"
193
8
	"launchpad.net/juju-core/state/testing"
194
9
	coretesting "launchpad.net/juju-core/testing"
195
10
	"launchpad.net/juju-core/worker/firewaller"
196
11
	"sort"
197
12
	"strings"
198
13
	stdtesting "testing"
199
14
	"time"
200
15
)
201
16
202
17
func TestPackage(t *stdtesting.T) {
203
18
	coretesting.ZkTestPackage(t)
204
19
}
205
20
206
21
// hooLogger allows the grabbing of debug log statements
207
22
// to compare them inside the tests.
208
23
type hookLogger struct {
209
24
	event     chan string
210
25
	oldTarget log.Logger
211
26
}
212
27
213
28
var logHook *hookLogger
214
29
215
30
const prefix = "JUJU:DEBUG firewaller: "
216
31
217
32
func (h *hookLogger) Output(calldepth int, s string) error {
218
33
	err := h.oldTarget.Output(calldepth, s)
219
34
	if strings.HasPrefix(s, prefix) {
220
35
		h.event <- s[len(prefix):]
221
36
	}
222
37
	return err
223
38
}
224
39
225
40
func setUpLogHook() {
226
41
	logHook = &hookLogger{
227
42
		event:     make(chan string, 30),
228
43
		oldTarget: log.Target,
229
44
	}
230
45
	log.Target = logHook
231
46
}
232
47
233
48
func tearDownLogHook() {
234
49
	log.Target = logHook.oldTarget
235
50
}
236
51
237
52
// assertEvents asserts that the expected events are received from
238
53
// the firewaller, in no particular order.
239
54
func assertEvents(c *C, expect []string) {
240
55
	var got []string
241
56
	for _ = range expect {
242
57
		select {
243
58
		case e := <-logHook.event:
244
59
			got = append(got, e)
245
60
		case <-time.After(500 * time.Millisecond):
246
61
			c.Fatalf("expected %q; timed out after %q", expect, got)
247
62
		}
248
63
	}
249
64
	select {
250
65
	case e := <-logHook.event:
251
66
		got = append(got, e)
252
67
		c.Fatalf("expected %q; too many events %q ", expect, got)
253
68
	case <-time.After(100 * time.Millisecond):
254
69
	}
255
70
	sort.Strings(expect)
256
71
	sort.Strings(got)
257
72
	c.Assert(got, DeepEquals, expect)
258
73
}
259
74
260
75
type FirewallerSuite struct {
261
76
	coretesting.LoggingSuite
262
77
	testing.StateSuite
263
78
	op <-chan dummy.Operation
264
79
}
265
80
266
81
var _ = Suite(&FirewallerSuite{})
267
82
268
83
func (s *FirewallerSuite) SetUpTest(c *C) {
269
84
	s.LoggingSuite.SetUpTest(c)
270
85
271
86
	op := make(chan dummy.Operation, 500)
272
87
	dummy.Listen(op)
273
88
	s.op = op
274
89
275
90
	s.StateSuite.SetUpTest(c)
276
91
}
277
92
278
93
func (s *FirewallerSuite) TearDownTest(c *C) {
279
94
	dummy.Reset()
280
95
	s.LoggingSuite.TearDownTest(c)
281
96
}
282
97
283
98
func (s *FirewallerSuite) TestStartStop(c *C) {
284
99
	fw, err := firewaller.NewFirewaller(s.State)
285
100
	c.Assert(err, IsNil)
286
101
	c.Assert(fw.Stop(), IsNil)
287
102
}
288
103
289
104
func (s *FirewallerSuite) TestAddRemoveMachine(c *C) {
290
105
	fw, err := firewaller.NewFirewaller(s.State)
291
106
	c.Assert(err, IsNil)
292
107
293
108
	setUpLogHook()
294
109
	defer tearDownLogHook()
295
110
296
111
	m1, err := s.State.AddMachine()
297
112
	c.Assert(err, IsNil)
298
113
	m2, err := s.State.AddMachine()
299
114
	c.Assert(err, IsNil)
300
115
	m3, err := s.State.AddMachine()
301
116
	c.Assert(err, IsNil)
302
117
303
118
	assertEvents(c, []string{
304
119
		fmt.Sprint("add-machine ", m1.Id()),
305
120
		fmt.Sprint("add-machine ", m2.Id()),
306
121
		fmt.Sprint("add-machine ", m3.Id()),
307
122
	})
308
123
309
124
	err = s.State.RemoveMachine(m2.Id())
310
125
	c.Assert(err, IsNil)
311
126
312
127
	assertEvents(c, []string{
313
128
		fmt.Sprint("remove-machine ", m2.Id()),
314
129
	})
315
130
316
131
	c.Assert(fw.Stop(), IsNil)
317
132
}
318
133
319
134
func (s *FirewallerSuite) TestFirewallerStopOnStateClose(c *C) {
320
135
	fw, err := firewaller.NewFirewaller(s.State)
321
136
	c.Assert(err, IsNil)
322
137
	fw.CloseState()
323
138
	c.Check(fw.Wait(), ErrorMatches, ".* zookeeper is closing")
324
139
	c.Assert(fw.Stop(), ErrorMatches, ".* zookeeper is closing")
325
140
}
Status:	Merged
Approved by:	Gustavo Niemeyer on 2012-07-19
Approved revision:	302
Merged at revision:	313
Proposed branch:	lp:~themue/juju-core/go-worker-firewaller-machines
Merge into:	lp:~juju/juju-core/trunk
Diff against target:	325 lines (+310/-0) 3 files modified worker/firewaller/export_test.go (+16/-0) worker/firewaller/firewaller.go (+154/-0) worker/firewaller/firewaller_test.go (+140/-0)
To merge this branch:	bzr merge lp:~themue/juju-core/go-worker-firewaller-machines
Related bugs:	Link a bug report
Reviewer	Review Type	Date Requested	Status
The Go Language Gophers		2012-07-13	Pending
Review via email: mp+114938@code.launchpad.net