Merge lp:~jimbaker/pyjuju/expose-provision-machines-reexpose into lp:pyjuju

[Jim Baker]

Fixing this was essentially a matter of making
cb_watch_service_exposed_flag be symmetric on opening/closing ports on
the corresponding service units; that is, the following code should be
run regardless of whether the service is exposed:

            for unit_state in unit_states:
                yield self.open_close_ports(unit_state)

Tests in test_provision have been changed so that they actually test
re-exposing a service.

In addition, the support for re-exposing brought up the issue that
StateBase._watch_topology has the same issue previously identified in
other watches: that it needs to guard on the connection being
connected upon entry to the watch function. These changes were made in
both _watch_topology and __watch_topology. Tests were also add to
test_base to verify that when the client is disconnected, the watch
properly shuts down. (The necessity for this change to StateBase is
otherwise only seen in repeated looping of tests that in the
provisioning agent.)

Lastly, I removed the unnecessary sleeps in test_base in favor of our
current practice of using poke_zk (but only for these types of tests).

[Gustavo Niemeyer (niemeyer)]

Looks good, thanks Jim.

review: Approve

[Kapil Thangavelu (hazmat)]

looks good, thanks for digging into the additional robustness checks.

review: Approve

[Jim Baker (jimbaker)]

Thanks. I will be merging this into trunk, along with its upstream,
expose-provision-machines, once expose-provider-ec2 is in place, since it
requires a provider implementation to be present.

On Mon, Jul 25, 2011 at 11:01 AM, Kapil Thangavelu <
<email address hidden>> wrote:

> The proposal to merge
> lp:~jimbaker/ensemble/expose-provision-machines-reexpose into lp:ensemble
> has been updated.
>
> Status: Needs review => Approved
>
> For more details, see:
>
> https://code.launchpad.net/~jimbaker/ensemble/expose-provision-machines-reexpose/+merge/68313<https://code.launchpad.net/%7Ejimbaker/ensemble/expose-provision-machines-reexpose/+merge/68313>
> --
>
> https://code.launchpad.net/~jimbaker/ensemble/expose-provision-machines-reexpose/+merge/68313<https://code.launchpad.net/%7Ejimbaker/ensemble/expose-provision-machines-reexpose/+merge/68313>
> You are the owner of
> lp:~jimbaker/ensemble/expose-provision-machines-reexpose.
>

[lp:~jimbaker/pyjuju/expose-provision-machines-reexpose]

301. By Jim Baker on 2011-07-27

Merged upstream branch expose-provision-machines

302. By Jim Baker on 2011-07-29

Merged upstream expose-provision-machines

303. By Jim Baker on 2011-07-29

Merged upstream expose-provision-machines

304. By Jim Baker on 2011-08-08

Merged upstream

305. By Jim Baker on 2011-08-09

Merged upstream

=== modified file 'ensemble/agents/provision.py'
--- ensemble/agents/provision.py	2011-08-09 14:27:56 +0000
+++ ensemble/agents/provision.py	2011-08-09 14:27:56 +0000
@@ -301,16 +301,23 @@
         def cb_watch_service_exposed_flag(exposed):
             if not self._running:
                 raise StopWatcher()
+
+            if exposed:
+                log.debug("Service %r is exposed", service_name)
+            else:
+                log.debug("Service %r is unexposed", service_name)
+
+            try:
+                unit_states = yield service_state.get_all_unit_states()
+            except StateChanged:
+                log.debug("Stopping watch on %r, no longer in topology",
+                          service_name)
+                raise StopWatcher()
+            for unit_state in unit_states:
+                yield self.open_close_ports(unit_state)
+
             if not exposed:
                 self._watched_services[service_name] = NotExposed
-                try:
-                    unit_states = yield service_state.get_all_unit_states()
-                except StateChanged:
-                    log.debug("Stopping watch on %r, no longer in topology",
-                             service_name)
-                    raise StopWatcher()
-                for unit_state in unit_states:
-                    yield self.open_close_ports(unit_state)
             else:
                 self._watched_services[service_name] = set()
                 yield self._setup_service_unit_watch(service_state)

=== modified file 'ensemble/agents/tests/test_provision.py'
--- ensemble/agents/tests/test_provision.py	2011-08-09 14:27:56 +0000
+++ ensemble/agents/tests/test_provision.py	2011-08-09 14:27:56 +0000
@@ -507,6 +507,14 @@
         yield wordpress.clear_exposed_flag()
         self.assertTrue((yield expected_units))
 
+        # Re-expose wordpress: set the flag again, verify that it
+        # triggers on the expected units
+        expected_units = self.wait_on_expected_units(
+            set(["wordpress/0"]))
+        yield wordpress.set_exposed_flag()
+        self.assertTrue((yield expected_units))
+        yield self.agent.stop()
+
     @inlineCallbacks
     def test_add_remove_service_units_for_exposed_service(self):
         """Verify that adding/removing service units for an exposed
@@ -799,8 +807,8 @@
         yield drupal_0.open_port(443, "tcp")
         yield wordpress_0.open_port(80, "tcp")
         yield wordpress_1.open_port(80, "tcp")
+        self.assertTrue((yield expected_units))
         self.assertTrue((yield expected_machines))
-        self.assertTrue((yield expected_units))
         self.assertEqual((yield self.get_provider_ports(machine_0)),
                          set([(80, "tcp"), (443, "tcp"), (8080, "tcp")]))
         self.assertEqual((yield self.get_provider_ports(machine_1)),
@@ -869,4 +877,22 @@
         self.assertTrue((yield expected_units))
         self.assertEqual((yield self.get_provider_ports(machine_0)),
                          set([(80, "tcp"), (443, "tcp"), (8080, "tcp")]))
+
+        # Unexpose drupal service, verify only wordpress ports are now opened
+        expected_machines = self.wait_on_expected_machines(set([0]))
+        expected_units = self.wait_on_expected_units(set(["drupal/0"]))
+        yield drupal.clear_exposed_flag()
+        self.assertTrue((yield expected_machines))
+        self.assertTrue((yield expected_units))
+        self.assertEqual((yield self.get_provider_ports(machine_0)),
+                         set([(80, "tcp")]))
+
+        # Re-expose drupal service, verify ports are once again opened
+        expected_machines = self.wait_on_expected_machines(set([0]))
+        expected_units = self.wait_on_expected_units(set(["drupal/0"]))
+        yield drupal.set_exposed_flag()
+        self.assertTrue((yield expected_machines))
+        self.assertTrue((yield expected_units))
+        self.assertEqual((yield self.get_provider_ports(machine_0)),
+                         set([(80, "tcp"), (443, "tcp"), (8080, "tcp")]))
         yield self.agent.stop()

=== modified file 'ensemble/state/base.py'
--- ensemble/state/base.py	2011-05-06 04:40:33 +0000
+++ ensemble/state/base.py	2011-08-09 14:27:56 +0000
@@ -94,6 +94,13 @@
         "private", since the only purpose of this method is to be used by
         subclasses.
         """
+        # Need to guard on the client being connected in the case
+        # 1) a watch is waiting to run (in the reactor);
+        # 2) and the connection is closed.
+        # Because _watch_topology always chains to __watch_topology,
+        # the other guarding seen with `StopWatcher` is done there.
+        if not self._client.connected:
+            return
         exists, watch = self._client.exists_and_watch("/topology")
         stat = yield exists
 
@@ -106,6 +113,14 @@
     @inlineCallbacks
     def __topology_changed(self, ignored, watch_topology_function):
         """Internal callback used by _watch_topology()."""
+        # Need to guard on the client being connected in the case
+        # 1) a watch is waiting to run (in the reactor);
+        # 2) and the connection is closed.
+        #
+        # It remains the reponsibility of `watch_topology_function` to
+        # raise `StopWatcher`, per the doc of `_topology_changed`.
+        if not self._client.connected:
+            return
         try:
             get, watch = self._client.get_and_watch("/topology")
             content, stat = yield get

=== modified file 'ensemble/state/tests/test_base.py'
--- ensemble/state/tests/test_base.py	2011-02-10 20:13:55 +0000
+++ ensemble/state/tests/test_base.py	2011-08-09 14:27:56 +0000
@@ -1,11 +1,12 @@
 from twisted.internet.defer import inlineCallbacks, Deferred
-from twisted.internet import reactor
+from txzookeeper import ZookeeperClient
 
 from ensemble.state.tests.common import StateTestBase
 
 from ensemble.state.base import StateBase
 from ensemble.state.errors import StopWatcher
 from ensemble.state.topology import InternalTopology
+from ensemble.tests.common import get_test_zookeeper_address
 
 
 class StateBaseTest(StateTestBase):
@@ -176,9 +177,7 @@
         yield wait_callback[1]
 
         # Give a chance for something bad to happen.
-        go_reactor = Deferred()
-        reactor.callLater(0.5, go_reactor.callback, None)
-        yield go_reactor
+        yield self.poke_zk()
 
         # Now the watch callback must have been fired with two
         # different topologies.  The old one, and the new one.
@@ -273,7 +272,7 @@
         yield self.set_topology(topology)
 
         # Give a chance for something bad to happen.
-        yield self.sleep(0.1)
+        yield self.poke_zk()
 
         # Ensure we still have a single call.
         self.assertEquals(len(calls), 1)
@@ -295,8 +294,8 @@
     @inlineCallbacks
     def test_stop_watch(self):
         """
-        A watch which fires an StopWatcher excepiton, will
-        end the watch."""
+        A watch that fires a `StopWatcher` exception will end the
+        watch."""
         wait_callback = [Deferred() for i in range(5)]
         calls = []
 
@@ -324,12 +323,115 @@
         yield wait_callback[1]
         self.assertEqual(len(calls), 2)
 
-        # Change the topology again, we shouldnt' see this.
+        # Change the topology again, we shouldn't see this.
         topology.add_machine("m-2")
         yield self.set_topology(topology)
 
         # Give a chance for something bad to happen.
-        yield self.sleep(0.1)
+        yield self.poke_zk()
 
         # Ensure we still have a single call.
         self.assertEquals(len(calls), 2)
+
+    @inlineCallbacks
+    def test_watch_stops_on_closed_connection(self):
+        """Verify watches stops when the connection is closed."""
+
+        # Use a separate client connection for watching so it can be
+        # disconnected.
+        watch_client = ZookeeperClient(get_test_zookeeper_address())
+        yield watch_client.connect()
+        watch_base = StateBase(watch_client)
+
+        wait_callback = Deferred()
+        finish_callback = Deferred()
+        calls = []
+
+        def watcher(old_topology, new_topology):
+            calls.append((old_topology, new_topology))
+            wait_callback.callback(True)
+            return finish_callback
+
+        # Start watching.
+        yield watch_base._watch_topology(watcher)
+
+        # Create the topology.
+        topology = InternalTopology()
+        topology.add_machine("m-0")
+        yield self.set_topology(topology)
+        
+        # Hold off until callback is started.
+        yield wait_callback
+
+        # Change the topology.
+        topology.add_machine("m-1")
+        yield self.set_topology(topology)
+
+        # Ensure that the watch has been called just once so far
+        # (although still pending due to the finish_callback).
+        self.assertEquals(len(calls), 1)
+
+        # Now disconnect the client.
+        watch_client.close()
+        self.assertFalse(watch_client.connected)
+        self.assertTrue(self.client.connected)
+
+        # Change the topology again.
+        topology.add_machine("m-2")
+        yield self.set_topology(topology)
+
+        # Allow the first call to be completed, starting a process of
+        # watching for the next change. At this point, the watch will
+        # encounter that the client is disconnected.
+        finish_callback.callback(True)
+
+        # Give a chance for something bad to happen.
+        yield self.poke_zk()
+
+        # Ensure the watch was still not called.
+        self.assertEquals(len(calls), 1)
+
+    @inlineCallbacks
+    def test_watch_stops_on_early_closed_connection(self):
+        """Verify watches stops when the connection is closed early.
+
+        _watch_topology chains from an exists_and_watch to a
+        get_and_watch. This test ensures that this chaining will fail
+        gracefully if the connection is closed before this chaining
+        can occur.
+        """
+        # Use a separate client connection for watching so it can be
+        # disconnected.
+        watch_client = ZookeeperClient(get_test_zookeeper_address())
+        yield watch_client.connect()
+        watch_base = StateBase(watch_client)
+
+        calls = []
+
+        @inlineCallbacks
+        def watcher(old_topology, new_topology):
+            calls.append((old_topology, new_topology))
+
+        # Create the topology.
+        topology = InternalTopology()
+        topology.add_machine("m-0")
+        yield self.set_topology(topology)
+        
+        # Now disconnect the client.
+        watch_client.close()
+        self.assertFalse(watch_client.connected)
+        self.assertTrue(self.client.connected)
+
+        # Start watching.
+        yield watch_base._watch_topology(watcher)
+
+        # Change the topology, this will trigger the watch.
+        topology.add_machine("m-1")
+        yield self.set_topology(topology)
+
+        # Give a chance for something bad to happen.
+        yield self.poke_zk()
+
+        # Ensure the watcher was never called, because its client was
+        # disconnected.
+        self.assertEquals(len(calls), 0)