Merge lp:~themue/juju-core/go-worker-firewaller-machineunits into lp:~juju/juju-core/trunk

[Frank Mueller]

firewaller: added handling of machine units changes

The second iteration of the firewaller adds the handling of
changes of the machine units. Here the adding of services is
missing and will be one of the next steps. The next step
itself will be the handling of unit ports changes.

https://codereview.appspot.com/6404051/

[Roger Peppe (rogpeppe)]

LGTM

https://codereview.appspot.com/6404051/diff/1/worker/firewaller/firewaller.go
File worker/firewaller/firewaller.go (right):

https://codereview.appspot.com/6404051/diff/1/worker/firewaller/firewaller.go#newcode101
worker/firewaller/firewaller.go:101: if change.stateChange != nil {
when could change.stateChange be nil?

https://codereview.appspot.com/6404051/diff/1/worker/firewaller/firewaller.go#newcode153
worker/firewaller/firewaller.go:153: func (m *machine) loop(fw
*Firewaller) {
perhaps a comment on this function?

// loop waits for changes to the machine's units
// and funnels them to the firewall's machineUnitsChanges
// channel.

https://codereview.appspot.com/6404051/

[William Reade (fwereade)]

Essentially, same issues as the prereq.

https://codereview.appspot.com/6404051/diff/1/worker/firewaller/firewaller.go
File worker/firewaller/firewaller.go (right):

https://codereview.appspot.com/6404051/diff/1/worker/firewaller/firewaller.go#newcode98
worker/firewaller/firewaller.go:98: fw.tomb.Killf("aggregation of
machine units changes failed")
This maybe also feels like a panic situation, now I come to look at it.
(I know we don't panic in every analogous situation, but maybe we
should: closure of a locally-owned channel tat should stay open until
the method exits feels like solid evidence of fundamental logic error to
me)

https://codereview.appspot.com/6404051/diff/1/worker/firewaller/firewaller.go#newcode106
worker/firewaller/firewaller.go:106: // a not yet managed unit?
+1

https://codereview.appspot.com/6404051/diff/1/worker/firewaller/firewaller.go#newcode108
worker/firewaller/firewaller.go:108: u.watcher.Stop()
Doesn't guarantee we won't get more events on unitPortsChanges (see
previous review).

https://codereview.appspot.com/6404051/diff/1/worker/firewaller/firewaller.go#newcode198
worker/firewaller/firewaller.go:198: return
fw.tomb.Kill(watcher.MustErr(u.watcher))

(uncertainty as in previous review)

https://codereview.appspot.com/6404051/

[Gustavo Niemeyer (niemeyer)]

There are unhandled reviews on this. I'll wait for the next round.

https://codereview.appspot.com/6404051/

[Roger Peppe (rogpeppe)]

LGTM

https://codereview.appspot.com/6404051/diff/11057/worker/firewaller/firewaller.go
File worker/firewaller/firewaller.go (right):

https://codereview.appspot.com/6404051/diff/11057/worker/firewaller/firewaller.go#newcode64
worker/firewaller/firewaller.go:64: case change, ok :=
<-fw.machineUnitsChanges:
we never close machineUnitsChanges (trivially verifiable), so why not
just:

case change := <-fw.machineUnitsChanges:

and lose the test?

https://codereview.appspot.com/6404051/diff/11057/worker/firewaller/firewaller.go#newcode69
worker/firewaller/firewaller.go:69: log.Printf("tracker of machine %d
terminated prematurely", change.machine.id)
we're losing the error message from the machine tracker termination
here. how about:
log.Printf("tracker of machine %d terminated prematurely: %v",
change.machine.stop())
?

https://codereview.appspot.com/6404051/diff/11057/worker/firewaller/firewaller.go#newcode136
worker/firewaller/firewaller.go:136: // newMachineTracker creates a new
machine tracker keeping track of
it occurs to me that this thing doesn't actually keep track of anything
- it simply forwards machine units changes to
the central goroutine.

perhaps:
// newMachineTracker tracks changes unit changes to
// the given machine and sends them to the
// central firewaller loop.

?

https://codereview.appspot.com/6404051/

[Gustavo Niemeyer (niemeyer)]

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go
File worker/firewaller/firewaller.go (right):

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode16
worker/firewaller/firewaller.go:16: machines
map[int]*machineTracker
s/machines/machineTrackers

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode18
worker/firewaller/firewaller.go:18: units
map[string]*unitTracker
s/units/unitTrackers/

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode20
worker/firewaller/firewaller.go:20: services
map[string]*serviceTracker
s/services/serviceTrackers

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode48
worker/firewaller/firewaller.go:48: for _, removedMachine := range
change.Removed {
s/removedMachine/machine/

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode49
worker/firewaller/firewaller.go:49: mt, ok :=
fw.machines[removedMachine.Id()]
s/mt/machinet/

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode59
worker/firewaller/firewaller.go:59: for _, addedMachine := range
change.Added {
s/addedMachine/machine/

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode60
worker/firewaller/firewaller.go:60: mt :=
newMachineTracker(addedMachine, fw)
s/mt/machinet/

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode65
worker/firewaller/firewaller.go:65: if change.change == nil {
As we talked, this seems to just create extra churn. I think we can drop
it, and change the sending side so it never sends nil stuff.

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode70
worker/firewaller/firewaller.go:70: for _, removedUnit := range
change.change.Removed {
s/removedUnit/unit/

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode71
worker/firewaller/firewaller.go:71: ut, ok :=
fw.units[removedUnit.Name()]
s/ut/unitt/

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode79
worker/firewaller/firewaller.go:79: log.Debugf("firewaller: stopped
tracking unit %s", removedUnit.Name())
Shouldn't the unit ports in this machine be closed here?

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode81
worker/firewaller/firewaller.go:81: for _, addedUnit := range
change.change.Added {
s/addedUnit/unit/

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode82
worker/firewaller/firewaller.go:82: ut := newUnitTracker(addedUnit, fw)
s/ut/unitt/

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode120
worker/firewaller/firewaller.go:120: machine *machineTracker
s/machine/machinet/

https://codereview.appspot.com/6404051/diff/13004/worker/firewaller/firewaller.go#newcode121
worker/firewaller/firewaller.go:121: change *state.MachineUnitsChange
s/change //; so that you can use the fields of the Change type directly
rather than doing change...

[Gustavo Niemeyer (niemeyer)]

LGTM

https://codereview.appspot.com/6404051/diff/10058/worker/firewaller/firewaller.go
File worker/firewaller/firewaller.go (right):

https://codereview.appspot.com/6404051/diff/10058/worker/firewaller/firewaller.go#newcode16
worker/firewaller/firewaller.go:16: machineDatas
map[int]*machineData
As discussed: machineds, unitds, serviceds

https://codereview.appspot.com/6404051/diff/10058/worker/firewaller/firewaller.go#newcode17
worker/firewaller/firewaller.go:17: machineUnitsChanges chan
*machineUnitsChange
s/machineUnitsChanges/unitsChange/ (both in the type name and in the
field name)

https://codereview.appspot.com/6404051/diff/10058/worker/firewaller/firewaller.go#newcode19
worker/firewaller/firewaller.go:19: unitPortsChanges chan
*unitPortsChange
s/unitPortsChanges/portsChange/ (both in the type and in the field name)

https://codereview.appspot.com/6404051/diff/10058/worker/firewaller/firewaller.go#newcode73
worker/firewaller/firewaller.go:73: log.Printf("unit data %s returned
error when stopping: %v", unit.Name(), err)
s/%s/%q/
s/data/watcher/

https://codereview.appspot.com/6404051/

[Gustavo Niemeyer (niemeyer)]

LGTM

https://codereview.appspot.com/6404051/

[lp:~themue/juju-core/go-worker-firewaller-machineunits]

308. By Frank Mueller on 2012-07-25

firewaller: forgot two name changes

309. By Frank Mueller on 2012-07-25

firewaller: merged trunk

=== modified file 'worker/firewaller/firewaller.go'
--- worker/firewaller/firewaller.go	2012-07-19 11:16:40 +0000
+++ worker/firewaller/firewaller.go	2012-07-25 08:25:37 +0000
@@ -10,20 +10,26 @@
 // Firewaller watches the state for ports opened or closed
 // and reflects those changes onto the backing environment.
 type Firewaller struct {
-	st                  *state.State
-	tomb                tomb.Tomb
-	machinesWatcher     *state.MachinesWatcher
-	machines            map[int]*machineTracker
-	machineUnitsChanges chan *machineUnitsChange
+	st              *state.State
+	tomb            tomb.Tomb
+	machinesWatcher *state.MachinesWatcher
+	machineds       map[int]*machineData
+	unitsChange     chan *machineUnitsChange
+	unitds          map[string]*unitData
+	portsChange     chan *unitPortsChange
+	serviceds       map[string]*serviceData
 }
 
 // NewFirewaller returns a new Firewaller.
 func NewFirewaller(st *state.State) (*Firewaller, error) {
 	fw := &Firewaller{
-		st:                  st,
-		machinesWatcher:     st.WatchMachines(),
-		machines:            make(map[int]*machineTracker),
-		machineUnitsChanges: make(chan *machineUnitsChange),
+		st:              st,
+		machinesWatcher: st.WatchMachines(),
+		machineds:       make(map[int]*machineData),
+		unitsChange:     make(chan *machineUnitsChange),
+		unitds:          make(map[string]*unitData),
+		portsChange:     make(chan *unitPortsChange),
+		serviceds:       make(map[string]*serviceData),
 	}
 	go fw.loop()
 	return fw, nil
@@ -39,24 +45,45 @@
 			if !ok {
 				return
 			}
-			for _, removedMachine := range change.Removed {
-				m, ok := fw.machines[removedMachine.Id()]
+			for _, machine := range change.Removed {
+				machined, ok := fw.machineds[machine.Id()]
 				if !ok {
 					panic("trying to remove machine that wasn't added")
 				}
-				delete(fw.machines, removedMachine.Id())
-				if err := m.stop(); err != nil {
-					log.Printf("machine tracker %d returned error when stopping: %v", removedMachine.Id(), err)
-				}
-				log.Debugf("firewaller: stopped tracking machine %d", removedMachine.Id())
-			}
-			for _, addedMachine := range change.Added {
-				m := newMachineTracker(addedMachine, fw)
-				fw.machines[addedMachine.Id()] = m
-				log.Debugf("firewaller: started tracking machine %d", m.id)
-			}
-		case <-fw.machineUnitsChanges:
-			// TODO(mue) fill with life.
+				delete(fw.machineds, machine.Id())
+				if err := machined.stopWatch(); err != nil {
+					log.Printf("machine data %d returned error when stopping: %v", machine.Id(), err)
+				}
+				log.Debugf("firewaller: stopped watching machine %d", machine.Id())
+			}
+			for _, machine := range change.Added {
+				machined := newMachineData(machine, fw)
+				fw.machineds[machine.Id()] = machined
+				log.Debugf("firewaller: started watching machine %d", machine.Id())
+			}
+		case change := <-fw.unitsChange:
+			for _, unit := range change.Removed {
+				unitd, ok := fw.unitds[unit.Name()]
+				if !ok {
+					panic("trying to remove unit that wasn't added")
+				}
+				delete(fw.unitds, unit.Name())
+				// TODO(mue) Close ports.
+				if err := unitd.stopWatch(); err != nil {
+					log.Printf("unit watcher %q returned error when stopping: %v", unit.Name(), err)
+				}
+				log.Debugf("firewaller: stopped watching unit %s", unit.Name())
+			}
+			for _, unit := range change.Added {
+				unitd := newUnitData(unit, fw)
+				fw.unitds[unit.Name()] = unitd
+				if fw.serviceds[unit.ServiceName()] == nil {
+					// TODO(mue) Add service watcher.
+				}
+				log.Debugf("firewaller: started watching unit %s", unit.Name())
+			}
+		case <-fw.portsChange:
+			// TODO(mue) Handle changes of ports.
 		}
 	}
 }
@@ -64,8 +91,11 @@
 // finishes cleans up when the firewaller is stopping.
 func (fw *Firewaller) finish() {
 	watcher.Stop(fw.machinesWatcher, &fw.tomb)
-	for _, m := range fw.machines {
-		fw.tomb.Kill(m.stop())
+	for _, unitd := range fw.unitds {
+		fw.tomb.Kill(unitd.stopWatch())
+	}
+	for _, machined := range fw.machineds {
+		fw.tomb.Kill(machined.stopWatch())
 	}
 	fw.tomb.Done()
 }
@@ -83,68 +113,118 @@
 
 // machineUnitsChange contains the changed units for one specific machine. 
 type machineUnitsChange struct {
-	machine *machineTracker
-	change  *state.MachineUnitsChange
+	machined *machineData
+	*state.MachineUnitsChange
 }
 
-// machineTracker keeps track of the unit changes of a machine.
-type machineTracker struct {
+// machineData watches the unit changes of a machine and passes them
+// to the firewaller for handling.
+type machineData struct {
 	tomb       tomb.Tomb
 	firewaller *Firewaller
-	id         int
+	machine    *state.Machine
 	watcher    *state.MachineUnitsWatcher
-	ports      map[state.Port]*unitTracker
-}
-
-// newMachineTracker creates a new machine tracker keeping track of
-// unit changes of the passed machine.
-func newMachineTracker(mst *state.Machine, fw *Firewaller) *machineTracker {
-	mt := &machineTracker{
-		firewaller: fw,
-		id:         mst.Id(),
-		watcher:    mst.WatchUnits(),
-		ports:      make(map[state.Port]*unitTracker),
-	}
-	go mt.loop()
-	return mt
-}
-
-// loop is the backend watching for machine units changes.
-func (mt *machineTracker) loop() {
-	defer mt.tomb.Done()
-	defer mt.watcher.Stop()
-	for {
-		select {
-		case <-mt.tomb.Dying():
-			return
-		case change, ok := <-mt.watcher.Changes():
-			// Send change or nil.
-			select {
-			case mt.firewaller.machineUnitsChanges <- &machineUnitsChange{mt, change}:
-			case <-mt.tomb.Dying():
-				return
-			}
-			// The watcher terminated prematurely, so end the loop.
-			if !ok {
-				mt.firewaller.tomb.Kill(watcher.MustErr(mt.watcher))
-				return
-			}
-		}
-	}
-}
-
-// stop stops the machine tracker.
-func (mt *machineTracker) stop() error {
-	mt.tomb.Kill(nil)
-	return mt.tomb.Wait()
-}
-
-type unitTracker struct {
-	service *serviceTracker
-	id      string
-	ports   []state.Port
-}
-
-type serviceTracker struct {
+}
+
+// newMachineData starts the watching of the passed machine. 
+func newMachineData(machine *state.Machine, fw *Firewaller) *machineData {
+	md := &machineData{
+		firewaller: fw,
+		machine:    machine,
+		watcher:    machine.WatchUnits(),
+	}
+	go md.watchLoop()
+	return md
+}
+
+// watchLoop is the backend watching for machine units changes.
+func (md *machineData) watchLoop() {
+	defer md.tomb.Done()
+	defer md.watcher.Stop()
+	for {
+		select {
+		case <-md.tomb.Dying():
+			return
+		case change, ok := <-md.watcher.Changes():
+			if !ok {
+				md.firewaller.tomb.Kill(watcher.MustErr(md.watcher))
+				return
+			}
+			select {
+			case md.firewaller.unitsChange <- &machineUnitsChange{md, change}:
+			case <-md.tomb.Dying():
+				return
+			}
+		}
+	}
+}
+
+// stopWatch stops the machine watching.
+func (md *machineData) stopWatch() error {
+	md.tomb.Kill(nil)
+	return md.tomb.Wait()
+}
+
+// unitPortsChange contains the changed ports for one specific unit. 
+type unitPortsChange struct {
+	unitd *unitData
+	ports []state.Port
+}
+
+// unitData watches the port changes of a unit and passes them
+// to the firewaller for handling.
+type unitData struct {
+	tomb       tomb.Tomb
+	firewaller *Firewaller
+	unit       *state.Unit
+	watcher    *state.PortsWatcher
+	service    *serviceData
+	ports      []state.Port
+}
+
+// newMachineData starts the watching of the passed unit. 
+func newUnitData(unit *state.Unit, fw *Firewaller) *unitData {
+	ud := &unitData{
+		firewaller: fw,
+		unit:       unit,
+		watcher:    unit.WatchPorts(),
+		ports:      make([]state.Port, 0),
+	}
+	go ud.watchLoop()
+	return ud
+}
+
+func (ud *unitData) watchLoop() {
+	defer ud.tomb.Done()
+	defer ud.watcher.Stop()
+	for {
+		select {
+		case <-ud.tomb.Dying():
+			return
+		case change, ok := <-ud.watcher.Changes():
+			if !ok {
+				ud.firewaller.tomb.Kill(watcher.MustErr(ud.watcher))
+				return
+			}
+			select {
+			case ud.firewaller.portsChange <- &unitPortsChange{ud, change}:
+			case <-ud.tomb.Dying():
+				return
+			}
+		}
+	}
+}
+
+// stopWatch stops the unit watching.
+func (ud *unitData) stopWatch() error {
+	ud.tomb.Kill(nil)
+	return ud.tomb.Wait()
+}
+
+// serviceData watches the exposed flag changes of a service and passes them
+// to the firewaller for handling.
+type serviceData struct {
+	// TODO(mue) Fill with life.
+	service *state.Service
 	exposed bool
 }

=== modified file 'worker/firewaller/firewaller_test.go'
--- worker/firewaller/firewaller_test.go	2012-07-19 11:16:40 +0000
+++ worker/firewaller/firewaller_test.go	2012-07-25 08:25:37 +0000
@@ -5,6 +5,7 @@
 	. "launchpad.net/gocheck"
 	"launchpad.net/juju-core/environs/dummy"
 	"launchpad.net/juju-core/log"
+	"launchpad.net/juju-core/state"
 	"launchpad.net/juju-core/state/testing"
 	coretesting "launchpad.net/juju-core/testing"
 	"launchpad.net/juju-core/worker/firewaller"
@@ -58,7 +59,7 @@
 		case e := <-logHook.event:
 			got = append(got, e)
 		case <-time.After(500 * time.Millisecond):
-			c.Fatalf("expected %q; timed out after %q", expect, got)
+			c.Fatalf("expected %q; timed out, got %q", expect, got)
 		}
 	}
 	select {
@@ -75,7 +76,8 @@
 type FirewallerSuite struct {
 	coretesting.LoggingSuite
 	testing.StateSuite
-	op <-chan dummy.Operation
+	op    <-chan dummy.Operation
+	charm *state.Charm
 }
 
 var _ = Suite(&FirewallerSuite{})
@@ -116,16 +118,56 @@
 	c.Assert(err, IsNil)
 
 	assertEvents(c, []string{
-		fmt.Sprint("started tracking machine ", m1.Id()),
-		fmt.Sprint("started tracking machine ", m2.Id()),
-		fmt.Sprint("started tracking machine ", m3.Id()),
+		fmt.Sprint("started watching machine ", m1.Id()),
+		fmt.Sprint("started watching machine ", m2.Id()),
+		fmt.Sprint("started watching machine ", m3.Id()),
 	})
 
 	err = s.State.RemoveMachine(m2.Id())
 	c.Assert(err, IsNil)
 
 	assertEvents(c, []string{
-		fmt.Sprint("stopped tracking machine ", m2.Id()),
+		fmt.Sprint("stopped watching machine ", m2.Id()),
+	})
+
+	c.Assert(fw.Stop(), IsNil)
+}
+
+func (s *FirewallerSuite) TestAssignUnassignUnit(c *C) {
+	fw, err := firewaller.NewFirewaller(s.State)
+	c.Assert(err, IsNil)
+
+	setUpLogHook()
+	defer tearDownLogHook()
+
+	m1, err := s.State.AddMachine()
+	c.Assert(err, IsNil)
+	m2, err := s.State.AddMachine()
+	c.Assert(err, IsNil)
+	s.charm = s.AddTestingCharm(c, "dummy")
+	s1, err := s.State.AddService("wordpress", s.charm)
+	c.Assert(err, IsNil)
+	u1, err := s1.AddUnit()
+	c.Assert(err, IsNil)
+	err = u1.AssignToMachine(m1)
+	c.Assert(err, IsNil)
+	u2, err := s1.AddUnit()
+	c.Assert(err, IsNil)
+	err = u2.AssignToMachine(m2)
+	c.Assert(err, IsNil)
+
+	assertEvents(c, []string{
+		fmt.Sprint("started watching machine ", m1.Id()),
+		fmt.Sprint("started watching machine ", m2.Id()),
+		fmt.Sprint("started watching unit ", u1.Name()),
+		fmt.Sprint("started watching unit ", u2.Name()),
+	})
+
+	err = u1.UnassignFromMachine()
+	c.Assert(err, IsNil)
+
+	assertEvents(c, []string{
+		fmt.Sprint("stopped watching unit ", u1.Name()),
 	})
 
 	c.Assert(fw.Stop(), IsNil)