Merge lp:~jimbaker/pyjuju/expose-cleanup into lp:pyjuju

[Jim Baker]

In the shutdown of EC2 machines, deletes the associated machine security groups (identified by using the Reservation metadata when describing instances), and if possible the environment security group.

This code necessarily does a polling of EC2, via describe_instances, to determine when the machine has moved from shutting-down to terminated, since only then can the security groups be deleted. In practice, the polling used here should be generous, but we may want to consider fine tuning the sleep interval (necessary to avoid unnecessarily hammering EC2) and the max number of polls. It might also be desirable to provide the latter as an option in the shutdown command.

[Gustavo Niemeyer (niemeyer)]

The logic looks mostly good, besides for [8] maybe. Some organizational issues, though.

[1]

+ def _get_ensemble_security_group(self):
+ return "ensemble-%s" % self.environment_name
+
+ def _get_machine_security_group(self, instance):
+ ensemble_security_group = self._get_ensemble_security_group()
+ for group in instance.reservation.groups:
+ if group != ensemble_security_group:
+ return group
+
+ # Ignore if no such group exists; this allows some limited
+ # backwards compatibility with old setups without machine
+ # security group
+ log.debug("No associated group for instance %r", instance.instance_id)
+ return None
+
+ @inlineCallbacks
+ def _delete_security_group(self, group):
+ yield self.ec2.delete_security_group(group)
+ log.debug("Deleted security group %r", group)

These utility functions also have little relation to the provider interface
itself. Please move them to top levels under ensemble.providers.ec2.securitygroup

[2]

+ # Repeatedly poll EC2 until instances are in terminated state;
(...)
+ except EC2Error:
+ pass # Ignore, cannot delete if other machines are using
+
+ returnValue(killable_machines)

Same thing. Big algorithm polluting the interface of the provider with
lots of domain-specific knowledge about a minor aspect.

Please create a function named remove_security_groups under the same
module as suggested above, and test it in isolation.

[3]

+ i = 0
+ while wait_on and i < 60:
(...)
+ i += 1
+ if wait_on:
+ yield self._sleep(1) # Reasonable sleep to continue polling

Minor hint:

    for i in range(60):
        (...)
        if not wait_on:
            break
        yield self._sleep(1)

[4]

Try dropping the sleep in that loop entirely. The describe_instances
operation is a natural limiter.

When you do that, how many iterations do you get in practice before
the operation completes, and how many seconds it took?

[5]

+ def _sleep(self, delay):
+ """Non-blocking sleep."""
+ from twisted.internet import reactor
+
+ deferred = Deferred()
+ reactor.callLater(delay, deferred.callback, None)
+ return deferred

If you dropped it in the point above, remove this. If not, please move
this to ensemble.lib.twistedutils and test it accordingly.

[6]

+ log.debug("%d Info on instance: %s, %s",
+ i, instance.instance_id, instance.instance_state)

Slightly better debug message:

    "[iteraction %d] Instance %s state: %s"

[7]

+ log.debug("Instance %r was terminated",
+ instance.instance_id)

Drop this.. the exact same information was just logged.

[8]

My understanding is that describe_instances would blow up if some of the
instances were not found. If that's the case, the algorithm won't work
in practice.

[9]

+ # Wait for the security groups deleted in this round
+ yield DeferredList(deletions)

Why? It's killing parallelism unnecessarily, it seems.

[10]

+ log.info("Shutdo...

review: Needs Fixing

[Jim Baker (jimbaker)]

On Fri, Aug 19, 2011 at 2:27 PM, Gustavo Niemeyer <email address hidden>wrote:

> Review: Needs Fixing
> The logic looks mostly good, besides for [8] maybe. Some organizational
> issues, though.
>
> [1]
>
> + def _get_ensemble_security_group(self):
> + return "ensemble-%s" % self.environment_name
> +
> + def _get_machine_security_group(self, instance):
> + ensemble_security_group = self._get_ensemble_security_group()
> + for group in instance.reservation.groups:
> + if group != ensemble_security_group:
> + return group
> +
> + # Ignore if no such group exists; this allows some limited
> + # backwards compatibility with old setups without machine
> + # security group
> + log.debug("No associated group for instance %r",
> instance.instance_id)
> + return None
> +
> + @inlineCallbacks
> + def _delete_security_group(self, group):
> + yield self.ec2.delete_security_group(group)
> + log.debug("Deleted security group %r", group)
>
> These utility functions also have little relation to the provider interface
> itself. Please move them to top levels under
> ensemble.providers.ec2.securitygroup
>

Moved

>
> [2]
>
> + # Repeatedly poll EC2 until instances are in terminated state;
> (...)
> + except EC2Error:
> + pass # Ignore, cannot delete if other machines are using
> +
> + returnValue(killable_machines)
>
> Same thing. Big algorithm polluting the interface of the provider with
> lots of domain-specific knowledge about a minor aspect.
>
> Please create a function named remove_security_groups under the same
> module as suggested above, and test it in isolation.
>

Moved and tested now also in test_securitygroup. Note there are still tests
in test_shutdown

> [3]
>
> + i = 0
> + while wait_on and i < 60:
> (...)
> + i += 1
> + if wait_on:
> + yield self._sleep(1) # Reasonable sleep to continue
> polling
>
> Minor hint:
>
> for i in range(60):
> (...)
> if not wait_on:
> break
> yield self._sleep(1)
>

Changed to this form

>
> [4]
>
> Try dropping the sleep in that loop entirely. The describe_instances
> operation is a natural limiter.
>

Removed the sleep

>
> When you do that, how many iterations do you get in practice before
> the operation completes, and how many seconds it took?
>

and empirically observed approx 500ms per op

> [5]
>
> + def _sleep(self, delay):
> + """Non-blocking sleep."""
> + from twisted.internet import reactor
> +
> + deferred = Deferred()
> + reactor.callLater(delay, deferred.callback, None)
> + return deferred
>
> If you dropped it in the point above, remove this. If not, please move
> this to ensemble.lib.twistedutils and test it accordingly.
>

Removed

>
> [6]
>
> + log.debug("%d Info on instance: %s, %s",
> + i, instance.instance_id,
> instance.instance_state)
>
> Slightly better debug message:
>
> "[iteraction %d] Instance %s state: %s"
>
>
This was not intended to be in the branch pushed, so removed

> [...

[William Reade (fwereade)]

Just a few notes:

[0]

Sorry, you've got a potentially nasty merge inbound :(. Give me a shout if there's anything I can help with.

[1]

The security_group methods on EC2's MachineProvider, which moved to securitygroup.py, haven't been deleted from MachineProvider yet (but they do seem to be unused now)

[2]

+ returnValue(killable_machines)

I think it would be slightly more technically correct (the best kind of correct!) to construct terminated_machines from the list of terminated_ids, since you have them available.

[3]

+def remove_security_groups(provider, killable_ids):

Consider renaming killable_ids to terminated_ids, or perhaps instance_ids -- I can think of arguments either way, but I don't think "killable" is the right word to describe them at this stage.

[4]

+ # Delete the security group for the environment as a whole.

I think it would be more sensible to override MachineProvider.destroy_environment() on ec2.MachineProvider, and do this in there, rather than trying and failing on shutdown of any other machine. Note that the base destroy_environment can raise, so you might want something like this:

    def destroy_environment(self):
        try:
            super(MachineProvider, self).destroy_environment()
        finally:
            destroy_environment_security_group(self)

...which would at least ensure we made a good-faith attempt to do everything required to destroy the full environment, although the current logging in the face of multiple errors is a bit nonexistent.

All that said, LGTM; I'm very happy to see a clean API again. Thanks, and +1.

review: Approve

[Gustavo Niemeyer (niemeyer)]

This looks good. A couple of comments on top of the changes:

[11]

You got [9] reversed, as I pointed out on IRC:

<niemeyer> jimbaker: I was pointing out that you don't have to wait for deletion to finish on all of the removed groups before running another round
<niemeyer> jimbaker: since it kills parallelism unnecessarily
<niemeyer> jimbaker: You can wait for them to finish after _all_ of the groups have been deleted
<niemeyer> jimbaker: Instead, you've moved backwards, and made it wait on each individual group's deletion

[12]

+def _get_ensemble_security_group(provider):
+ """Get EC2 security group name for environment of `provider`."""
+ return "ensemble-%s" % provider.environment_name

+ def _get_ensemble_security_group(self):
+ return "ensemble-%s" % self.environment_name

There's no point in having that in a function if it's going to be
duplicated either way. Either unify, or just inline.

review: Approve

[lp:~jimbaker/pyjuju/expose-cleanup]

332. By Jim Baker on 2011-08-27

Delete security groups in the background

333. By Jim Baker on 2011-08-27

Removed debugging

334. By Jim Baker on 2011-08-28

Minor cleanup

335. By Jim Baker on 2011-08-28

Only remove env security group in context of provider destroy_environment

336. By Jim Baker on 2011-08-28

Reworked remaining mocks

337. By Jim Baker on 2011-08-29

Test new securitygroup function

338. By Jim Baker on 2011-08-29

Merged trunk & resolved conflicts

339. By Jim Baker on 2011-08-29

Fixed merge issues

340. By Jim Baker on 2011-08-29

Merged trunk

341. By Jim Baker on 2011-08-29

PEP8 & PyFlakes

[Jim Baker (jimbaker)]

On Wed, Aug 24, 2011 at 2:20 AM, William Reade
<email address hidden>wrote:

> Review: Approve
> Just a few notes:
>
> [0]
>
> Sorry, you've got a potentially nasty merge inbound :(. Give me a shout if
> there's anything I can help with.
>

It wasn't so bad, but definitely it wasn't trivial.

>
> [1]
>
> The security_group methods on EC2's MachineProvider, which moved to
> securitygroup.py, haven't been deleted from MachineProvider yet (but they do
> seem to be unused now)
>

Removed these (Gustavo pointed out one in his review too)

>
> [2]
>
> + returnValue(killable_machines)
>
> I think it would be slightly more technically correct (the best kind of
> correct!) to construct terminated_machines from the list of terminated_ids,
> since you have them available.
>

 Sorry, that was going to make it, but I just realized it wasn't done. I
could do that as a trivial.

> [3]
>
> +def remove_security_groups(provider, killable_ids):
>
> Consider renaming killable_ids to terminated_ids, or perhaps instance_ids
> -- I can think of arguments either way, but I don't think "killable" is the
> right word to describe them at this stage.
>

Renamed.

>
> [4]
>
> + # Delete the security group for the environment as a whole.
>
> I think it would be more sensible to override
> MachineProvider.destroy_environment() on ec2.MachineProvider, and do this in
> there, rather than trying and failing on shutdown of any other machine. Note
> that the base destroy_environment can raise, so you might want something
> like this:
>
> def destroy_environment(self):
> try:
> super(MachineProvider, self).destroy_environment()
> finally:
> destroy_environment_security_group(self)
>
> ...which would at least ensure we made a good-faith attempt to do
> everything required to destroy the full environment, although the current
> logging in the face of multiple errors is a bit nonexistent.
>

Changed as requested and a number of tests suitably adjusted in their
mocking expectations.

[Jim Baker (jimbaker)]

On Thu, Aug 25, 2011 at 2:24 PM, Gustavo Niemeyer <email address hidden>wrote:

> Review: Approve
> This looks good. A couple of comments on top of the changes:
>
> [11]
>
> You got [9] reversed, as I pointed out on IRC:
>
> <niemeyer> jimbaker: I was pointing out that you don't have to wait for
> deletion to finish on all of the removed groups before running another round
> <niemeyer> jimbaker: since it kills parallelism unnecessarily
> <niemeyer> jimbaker: You can wait for them to finish after _all_ of the
> groups have been deleted
> <niemeyer> jimbaker: Instead, you've moved backwards, and made it wait on
> each individual group's deletion
>

Machine security groups are now deleted in the background. The environment
security group deletion is performed in a separate step, due to William's
point that should be done from destroy_environment, which made a lot of
sense.

>
> [12]
>
>
> +def _get_ensemble_security_group(provider):
> + """Get EC2 security group name for environment of `provider`."""
> + return "ensemble-%s" % provider.environment_name
>
> + def _get_ensemble_security_group(self):
> + return "ensemble-%s" % self.environment_name
>
> There's no point in having that in a function if it's going to be
> duplicated either way. Either unify, or just inline.
>
> Removed, this was code that wasn't deleted as part of the move to
securitygroup.py (as in a move operation can be thought of as add + delete
:) ). William noticed the same thing.

=== modified file 'ensemble/control/shutdown.py'
--- ensemble/control/shutdown.py	2011-08-11 19:49:32 +0000
+++ ensemble/control/shutdown.py	2011-08-29 18:36:25 +0000
@@ -22,7 +22,8 @@
 
     value = raw_input(
         "Warning, this will destroy all machines and services in the\n"
-        "environment. Continue [y/N]")
+        "%r (type: %s) environment. Continue [y/N]" % (
+        environment.name, environment.type))
 
     if value.strip().lower() not in ("y", "yes"):
         options.log.info("Shutdown aborted")

=== modified file 'ensemble/providers/ec2/__init__.py'
--- ensemble/providers/ec2/__init__.py	2011-08-26 12:24:00 +0000
+++ ensemble/providers/ec2/__init__.py	2011-08-29 18:36:25 +0000
@@ -2,7 +2,6 @@
 import re
 
 from twisted.internet.defer import inlineCallbacks, returnValue
-
 from txaws.ec2.exception import EC2Error
 from txaws.service import AWSServiceRegion
 
@@ -14,7 +13,8 @@
 from .launch import EC2LaunchMachine
 from .machine import EC2ProviderMachine, machine_from_instance
 from .securitygroup import (
-    open_provider_port, close_provider_port, get_provider_opened_ports)
+    open_provider_port, close_provider_port, get_provider_opened_ports,
+    remove_security_groups, destroy_environment_security_group)
 from .utils import get_region_uri
 
 
@@ -104,6 +104,21 @@
         returnValue(machines)
 
     @inlineCallbacks
+    def destroy_environment(self):
+        """Terminate all associated machines and security groups.
+
+        The super defintion of this method terminates each machine in
+        the environment; this needs to be augmented here by also
+        removing the security group for the environment.
+        """
+        try:
+            killed_machines = yield super(MachineProvider, self).\
+                destroy_environment()
+            returnValue(killed_machines)
+        finally:
+            yield destroy_environment_security_group(self)
+
+    @inlineCallbacks
     def shutdown_machines(self, machines):
         """Terminate machines associated with this provider.
 
@@ -119,9 +134,19 @@
 
         ids = [m.instance_id for m in machines]
         killable_machines = yield self.get_machines(ids)
-        if killable_machines:
-            killable_ids = [m.instance_id for m in killable_machines]
-            yield self.ec2.terminate_instances(*killable_ids)
+        if not killable_machines:
+            returnValue([])  # Nothing to do
+
+        killable_ids = [m.instance_id for m in killable_machines]
+        terminated = yield self.ec2.terminate_instances(*killable_ids)
+
+        # Pass on what was actually terminated, in the case the
+        # machine has somehow disappeared since get_machines
+        # above. This is to avoid getting EC2Error: Error Message:
+        # Invalid id when running ec2.describe_instances in
+        # remove_security_groups
+        terminated_ids = [info[0] for info in terminated]
+        yield remove_security_groups(self, terminated_ids)
         returnValue(killable_machines)
 
     def open_port(self, machine, machine_id, port, protocol="tcp"):

=== modified file 'ensemble/providers/ec2/machine.py'
--- ensemble/providers/ec2/machine.py	2011-08-17 13:34:45 +0000
+++ ensemble/providers/ec2/machine.py	2011-08-29 18:36:25 +0000
@@ -1,5 +1,3 @@
-import datetime
-
 from ensemble.machine import ProviderMachine
 
 

=== modified file 'ensemble/providers/ec2/securitygroup.py'
--- ensemble/providers/ec2/securitygroup.py	2011-08-23 09:59:59 +0000
+++ ensemble/providers/ec2/securitygroup.py	2011-08-29 18:36:25 +0000
@@ -1,11 +1,17 @@
-from twisted.internet.defer import inlineCallbacks, returnValue
+from twisted.internet.defer import Deferred, inlineCallbacks, returnValue
 from txaws.ec2.exception import EC2Error
 
 from ensemble.errors import ProviderInteractionError
+from ensemble.lib.twistutils import gather_results
 
 from .utils import log
 
 
+def _get_ensemble_security_group(provider):
+    """Get EC2 security group name for environment of `provider`."""
+    return "ensemble-%s" % provider.environment_name
+
+
 def _get_machine_group_name(provider, machine_id):
     """Get EC2 security group name associated just with `machine_id`."""
     return "ensemble-%s-%s" % (provider.environment_name, machine_id)
@@ -87,3 +93,106 @@
             # Ensemble (at this time at least)
             opened_ports.add((from_port, ip_permission.ip_protocol))
     returnValue(opened_ports)
+
+
+def _get_machine_security_group_from_instance(provider, instance):
+    """Parses the `reservation` of `instance` to get assoc machine group."""
+    ensemble_security_group = _get_ensemble_security_group(provider)
+    for group in instance.reservation.groups:
+        if group != ensemble_security_group:
+            return group
+
+    # Ignore if no such group exists; this allows some limited
+    # backwards compatibility with old setups without machine
+    # security group
+    log.info("Ignoring missing machine security group for instance %r",
+             instance.instance_id)
+    return None
+
+
+@inlineCallbacks
+def _delete_security_group(provider, group, completed, deletion_required=True):
+    """Wrap EC2 delete_security_group."""
+    try:
+        yield provider.ec2.delete_security_group(group)
+        log.debug("Deleted security group %r", group)
+        completed.callback(None)
+    except EC2Error, e:
+        if deletion_required:
+            completed.errback(ProviderInteractionError(
+                    "EC2 error when attempting to delete group %s: %s" % (
+                        group, e)))
+        else:
+            completed.callback(None)
+
+
+def _background_delete_security_group(provider, group, deletion_required=True):
+    """Runs deletion task in the background.
+
+    Wait on this returned deferred for completion of the deletion.
+    """
+    from twisted.internet import reactor
+
+    completed = Deferred()
+    reactor.callLater(
+        0, _delete_security_group,
+        provider, group, completed, deletion_required=deletion_required)
+    return completed
+
+
+@inlineCallbacks
+def remove_security_groups(provider, instance_ids):
+    """Remove security groups associated with `instance_ids` for `provider`"""
+    log.info(
+        "Waiting on %d EC2 instances to transition to terminated state, "
+        "this may take a while", len(instance_ids))
+
+    # Repeatedly poll EC2 until instances are in terminated state;
+    # upon reaching that state, delete associated machine security
+    # groups. The limit of 200 polls is arbitrary and could be
+    # specified by a command line option (and/or an overall
+    # timeout). It's based on an observed ~500 ms roundtrip time per
+    # call of the describe_instances web service, along with typically
+    # taking about 40s to move all instances to a terminated state.
+    wait_on = set(instance_ids)
+    pending_deletions = []
+    for i in xrange(200):
+        if not wait_on:
+            break
+        instances = yield provider.ec2.describe_instances(*wait_on)
+        for instance in instances:
+            if instance.instance_state == "terminated":
+                log.debug("Instance %r was terminated",
+                          instance.instance_id)
+                wait_on.discard(instance.instance_id)
+                group = _get_machine_security_group_from_instance(
+                    provider, instance)
+                if group:
+                    pending_deletions.append(
+                        _background_delete_security_group(provider, group))
+    if wait_on:
+        outstanding = [
+            _get_machine_security_group_from_instance(provider, instance)
+                for instance in instances]
+        log.error("Instance shutdown taking too long, "
+                  "could not delete groups %s",
+                 ", ".join(sorted(outstanding)))
+
+    # Wait for all pending deletions to complete
+    yield gather_results(pending_deletions)
+
+
+@inlineCallbacks
+def destroy_environment_security_group(provider):
+    """Delete the security group for the environment of `provider`"""
+    group = _get_ensemble_security_group(provider)
+    try:
+        yield provider.ec2.delete_security_group(group)
+        log.debug("Deleted environment security group %r", group)
+        returnValue(True)
+    except EC2Error, e:
+        # Ignore, since this is only attempting to cleanup
+        log.debug(
+            "Ignoring EC2 error when attempting to delete group %s: %s" % (
+                group, e))
+        returnValue(False)

=== modified file 'ensemble/providers/ec2/tests/common.py'
--- ensemble/providers/ec2/tests/common.py	2011-08-23 13:01:13 +0000
+++ ensemble/providers/ec2/tests/common.py	2011-08-29 18:36:25 +0000
@@ -14,6 +14,7 @@
 
 
 MATCH_AMI = MATCH(lambda image_id: image_id.startswith("ami-"))
+MATCH_GROUP = MATCH(lambda x: x.startswith("ensemble-moon"))
 
 
 class EC2TestMixin(object):
@@ -36,8 +37,11 @@
         """
         return MachineProvider(self.env_name, self.get_config())
 
-    def get_instance(self, instance_id, state="running", **kwargs):
-        groups = kwargs.pop("groups", ["ensemble-%s" % self.env_name])
+    def get_instance(self,
+                     instance_id, state="running", machine_id=42, **kwargs):
+        groups = kwargs.pop("groups",
+                            ["ensemble-%s" % self.env_name,
+                             "ensemble-%s-%s" % (self.env_name, machine_id)])
         reservation = Reservation("x", "y", groups=groups)
         return Instance(instance_id, state, reservation=reservation, **kwargs)
 
@@ -168,3 +172,45 @@
             # connect grabs the first host of a set.
             self.ec2.describe_instances(hosts[0].instance_id)
             self.mocker.result(succeed([hosts[0]]))
+
+
+class MockInstanceState(object):
+    """Mock the result of ec2_describe_instances when called successively.
+
+    Each call of :method:`get_round` returns a list of mock `Instance`
+    objects, using the state for that round. Instance IDs not used in
+    the round (and passed in from ec2_describe_instances) are
+    automatically skipped."""
+
+    def __init__(self, tester, instance_ids, machine_ids, states):
+        self.tester = tester
+        self.instance_ids = instance_ids
+        self.machine_ids = machine_ids
+        self.states = states
+        self.round = 0
+
+    def get_round(self, *current_instance_ids):
+        result = []
+        for instance_id, machine_id, state in zip(
+                self.instance_ids, self.machine_ids, self.states[self.round]):
+            if instance_id not in current_instance_ids:
+                # Ignore instance_ids that are no longer being
+                # described, because they have since moved into a
+                # terminated state
+                continue
+            result.append(self.tester.get_instance(instance_id,
+                                                   machine_id=machine_id,
+                                                   state=state))
+        self.round += 1
+        return succeed(result)
+
+
+class Observed(object):
+    """Minimal wrapper just to ensure :method:`add` returns a `Deferred`."""
+
+    def __init__(self):
+        self.items = set()
+
+    def add(self, item):
+        self.items.add(item)
+        return succeed(True)

=== modified file 'ensemble/providers/ec2/tests/test_connect.py'
--- ensemble/providers/ec2/tests/test_connect.py	2011-08-19 16:21:13 +0000
+++ ensemble/providers/ec2/tests/test_connect.py	2011-08-29 18:36:25 +0000
@@ -113,7 +113,7 @@
         # Hand back a real connected client to test the wait on initialization.
         instance = self.get_instance("i-foobar", dns_name="foo.example.com")
         connected_client = ZookeeperClient()
-        self.client = connected_client # for poke_zk
+        self.client = connected_client  # for poke_zk
         self.mock_connect(False, instance, succeed(connected_client))
 
         self.mocker.replay()

=== modified file 'ensemble/providers/ec2/tests/test_provider.py'
--- ensemble/providers/ec2/tests/test_provider.py	2011-08-18 12:19:22 +0000
+++ ensemble/providers/ec2/tests/test_provider.py	2011-08-29 18:36:25 +0000
@@ -115,6 +115,7 @@
         serialized.pop("authorized-keys", None)
         self.assertEqual(config, serialized)
 
+
 class FailCreateTest(TestCase):
 
     def test_conflicting_authorized_keys_options(self):

=== modified file 'ensemble/providers/ec2/tests/test_securitygroup.py'
--- ensemble/providers/ec2/tests/test_securitygroup.py	2011-08-18 12:19:22 +0000
+++ ensemble/providers/ec2/tests/test_securitygroup.py	2011-08-29 18:36:25 +0000
@@ -7,12 +7,13 @@
 from ensemble.lib.testing import TestCase
 from ensemble.machine import ProviderMachine
 from ensemble.providers.ec2.securitygroup import (
-    open_provider_port, close_provider_port, get_provider_opened_ports)
-
-from .common import EC2TestMixin
-
-
-class EC2SecurityGroupTest(EC2TestMixin, TestCase):
+    open_provider_port, close_provider_port, get_provider_opened_ports,
+    remove_security_groups, destroy_environment_security_group)
+from ensemble.providers.ec2.tests.common import (
+    EC2TestMixin, MATCH_GROUP, Observed, MockInstanceState)
+
+
+class EC2PortMgmtTest(EC2TestMixin, TestCase):
 
     @inlineCallbacks
     def test_open_provider_port(self):
@@ -127,3 +128,174 @@
             str(ex),
             "Unexpected EC2Error getting open ports on machine i-foobar: "
             "The instance ID 'i-foobar' does not exist")
+
+
+class EC2RemoveGroupsTest(EC2TestMixin, TestCase):
+
+    @inlineCallbacks
+    def test_remove_security_groups(self):
+        """Test normal running seen in polling instances for their state."""
+        system_state = MockInstanceState(
+            self,
+            ["i-amkillable",   "i-amkillabletoo"], [0, 2],
+            [["shutting-down", "shutting-down"],
+             ["shutting-down", "shutting-down"],
+             ["shutting-down", "terminated"],
+             ["terminated"]])
+
+        self.ec2.describe_instances("i-amkillable", "i-amkillabletoo")
+        self.mocker.call(system_state.get_round)
+        self.mocker.count(3)
+
+        self.ec2.describe_instances("i-amkillable")
+        self.mocker.call(system_state.get_round)
+
+        self.ec2.delete_security_group(MATCH_GROUP)
+        deleted_groups = Observed()
+        self.mocker.call(deleted_groups.add)
+        self.mocker.count(2)
+        self.mocker.replay()
+
+        provider = self.get_provider()
+        yield remove_security_groups(
+            provider, ["i-amkillable", "i-amkillabletoo"])
+        self.assertEquals(
+            deleted_groups.items,
+            set(["ensemble-moon-0", "ensemble-moon-2"]))
+
+    @inlineCallbacks
+    def test_machine_not_in_machine_security_group(self):
+        """Old environments do not have machines in security groups.
+
+        TODO It might be desirable to change this behavior to log as an
+        error, or even raise as a provider exception.
+        """
+        log = self.capture_logging()
+
+        # Instance is not in a machine security group, just its env
+        # security group (ensemble-moon)
+        self.ec2.describe_instances("i-amkillable")
+        self.mocker.result(succeed([
+            self.get_instance("i-amkillable", "terminated", groups=[])]))
+        self.mocker.replay()
+
+        provider = self.get_provider()
+        yield remove_security_groups(provider, ["i-amkillable"])
+        self.assertIn(
+            "Ignoring missing machine security group for instance "
+            "'i-amkillable'",
+            log.getvalue())
+
+    @inlineCallbacks
+    def test_cannot_delete_machine_security_group(self):
+        """Verify this raises error.
+
+        Note it is possible for a malicious admin to start other
+        machines with an existing machine security group, so need to
+        test this removal is robust in this case. Another scenario
+        that might cause this is that there's an EC2 timeout.
+        """
+        system_state = MockInstanceState(
+            self, ["i-amkillable"], [0],
+            [["shutting-down"],
+             ["shutting-down"],
+             ["shutting-down"],
+             ["terminated"]])
+
+        self.ec2.describe_instances("i-amkillable")
+        self.mocker.call(system_state.get_round)
+        self.mocker.count(4)
+
+        self.ec2.delete_security_group("ensemble-moon-0")
+        self.mocker.result(fail(
+                self.get_ec2_error(
+                    "ensemble-moon-0",
+                    format="There are active instances using security group %r"
+                    )))
+
+        self.mocker.replay()
+
+        provider = self.get_provider()
+        ex = yield self.assertFailure(
+            remove_security_groups(provider, ["i-amkillable"]),
+            ProviderInteractionError)
+        self.assertEquals(
+            str(ex),
+            "EC2 error when attempting to delete group ensemble-moon-0: "
+            "Error Message: There are active instances using security group "
+            "'ensemble-moon-0'")
+
+    @inlineCallbacks
+    def test_remove_security_groups_takes_too_long(self):
+        """Verify that removing security groups doesn't continue indefinitely.
+
+        This scenario might happen if the instance is taking too long
+        to report as going into the terminated state, although not
+        likely after 1000 polls!
+        """
+        log = self.capture_logging()
+        states = [["shutting-down", "shutting-down"],
+                  ["shutting-down", "shutting-down"],
+                  ["shutting-down", "terminated"]]
+        keeps_on_going = [["shutting-down"] for i in xrange(1000)]
+        states.extend(keeps_on_going)
+
+        system_state = MockInstanceState(
+            self, ["i-amkillable", "i-amkillabletoo"], [0, 2], states)
+
+        self.ec2.describe_instances("i-amkillable", "i-amkillabletoo")
+        self.mocker.call(system_state.get_round)
+        self.mocker.count(3)
+
+        # Keep the rounds going until it exits with the error message
+        self.ec2.describe_instances("i-amkillable")
+        self.mocker.call(system_state.get_round)
+        self.mocker.count(197)
+
+        # Verify that at least this machine security group was deleted
+        deleted_groups = Observed()
+        self.ec2.delete_security_group("ensemble-moon-2")
+        self.mocker.call(deleted_groups.add)
+
+        self.mocker.replay()
+
+        provider = self.get_provider()
+        yield remove_security_groups(
+            provider, ["i-amkillable", "i-amkillabletoo"])
+        self.assertEquals(deleted_groups.items, set(["ensemble-moon-2"]))
+        self.assertIn(
+            "Instance shutdown taking too long, "
+            "could not delete groups ensemble-moon-0",
+            log.getvalue())
+
+    @inlineCallbacks
+    def test_destroy_environment_security_group(self):
+        """Verify the deletion of the security group for the environment"""
+        self.ec2.delete_security_group("ensemble-moon")
+        self.mocker.result(succeed(True))
+        self.mocker.replay()
+
+        provider = self.get_provider()
+        destroyed = yield destroy_environment_security_group(provider)
+        self.assertTrue(destroyed)
+
+    @inlineCallbacks
+    def test_destroy_environment_security_group_missing(self):
+        """Verify ignores errors in deleting the env security group"""
+        log = self.capture_logging(level=logging.DEBUG)
+        self.ec2.delete_security_group("ensemble-moon")
+        self.mocker.result(fail(
+                self.get_ec2_error(
+                    "ensemble-moon",
+                    format="The security group %r does not exist"
+                    )))
+        self.mocker.replay()
+
+        provider = self.get_provider()
+        destroyed = yield destroy_environment_security_group(provider)
+        self.assertFalse(destroyed)
+        self.assertIn(
+            "Ignoring EC2 error when attempting to delete group "
+            "ensemble-moon: Error Message: The security group "
+            "'ensemble-moon' does not exist",
+            log.getvalue())

=== modified file 'ensemble/providers/ec2/tests/test_shutdown.py'
--- ensemble/providers/ec2/tests/test_shutdown.py	2011-08-16 08:33:47 +0000
+++ ensemble/providers/ec2/tests/test_shutdown.py	2011-08-29 18:36:25 +0000
@@ -1,7 +1,8 @@
-from twisted.internet.defer import fail, succeed
+from twisted.internet.defer import succeed, fail, inlineCallbacks
 
 from ensemble.lib.testing import TestCase
-from ensemble.providers.ec2.tests.common import EC2TestMixin
+from ensemble.providers.ec2.tests.common import (
+    EC2TestMixin, MATCH_GROUP, Observed, MockInstanceState)
 
 from ensemble.machine import ProviderMachine
 
@@ -15,23 +16,36 @@
 
 class EC2ShutdownMachineTest(EC2TestMixin, TestCase):
 
+    @inlineCallbacks
     def test_shutdown_machine(self):
-        instance = self.get_instance("i-foobar")
+        system_state = MockInstanceState(
+            self, ["i-foobar"], [0],
+            [["running"],
+             ["shutting-down"],
+             ["shutting-down"],
+             ["shutting-down"],
+             ["terminated"]])
+
         self.ec2.describe_instances("i-foobar")
-        self.mocker.result(succeed([instance]))
+        self.mocker.call(system_state.get_round)
+        self.mocker.count(5)
+
         self.ec2.terminate_instances("i-foobar")
         self.mocker.result(succeed([("i-foobar", "running", "shutting-down")]))
+
+        self.ec2.delete_security_group("ensemble-moon-0")
+        deleted_groups = Observed()
+        self.mocker.call(deleted_groups.add)
+
         self.mocker.replay()
 
         machine = EC2ProviderMachine("i-foobar")
         provider = self.get_provider()
-        d = provider.shutdown_machine(machine)
-
-        def verify(machine):
-            self.assertTrue(isinstance(machine, EC2ProviderMachine))
-            self.assertEquals(machine.instance_id, "i-foobar")
-        d.addCallback(verify)
-        return d
+        machine = yield provider.shutdown_machine(machine)
+        self.assertTrue(isinstance(machine, EC2ProviderMachine))
+        self.assertEquals(machine.instance_id, "i-foobar")
+        # Notably, ensemble-moon is not mock deleted
+        self.assertEquals(deleted_groups.items, set(["ensemble-moon-0"]))
 
     def test_shutdown_machine_invalid_group(self):
         """
@@ -56,7 +70,7 @@
     def test_shutdown_machine_invalid_machine(self):
         """
         Attempting to shutdown a machine that from a different provider
-        type will raise a syntaxerror.
+        type will raise a `ProviderError`.
         """
         self.mocker.replay()
         machine = ProviderMachine("i-foobar")
@@ -71,16 +85,14 @@
         d.addCallback(check_error)
         return d
 
+    @inlineCallbacks
     def test_shutdown_machines_none(self):
         self.mocker.replay()
         provider = self.get_provider()
-        d = provider.shutdown_machines([])
-
-        def verify(result):
-            self.assertEquals(result, [])
-        d.addCallback(verify)
-        return d
-
+        result = yield provider.shutdown_machines([])
+        self.assertEquals(result, [])
+
+    @inlineCallbacks
     def test_shutdown_machines_some_invalid(self):
         self.ec2.describe_instances("i-amkillable", "i-amalien", "i-amdead")
         self.mocker.result(succeed([
@@ -90,104 +102,148 @@
         self.mocker.replay()
 
         provider = self.get_provider()
-        d = provider.shutdown_machines([
-            EC2ProviderMachine("i-amkillable"),
-            EC2ProviderMachine("i-amalien"),
-            EC2ProviderMachine("i-amdead")])
-        self.failUnlessFailure(d, MachinesNotFound)
-
-        def verify(error):
-            self.assertEquals(str(error),
-                              "Cannot find machines: i-amalien, i-amdead")
-        d.addCallback(verify)
-        return d
-
+        ex = yield self.assertFailure(
+            provider.shutdown_machines([
+                    EC2ProviderMachine("i-amkillable"),
+                    EC2ProviderMachine("i-amalien"),
+                    EC2ProviderMachine("i-amdead")]),
+            MachinesNotFound)
+        self.assertEquals(str(ex),
+                          "Cannot find machines: i-amalien, i-amdead")
+
+    @inlineCallbacks
     def test_shutdown_machines_some_success(self):
+        """Verify that shutting down some machines works.
+
+        In particular, the environment as a whole is not removed
+        because there's still the environment security group left."""
+        system_state = MockInstanceState(
+            self,
+            ["i-amkillable",   "i-amkillabletoo"], [0, 2],
+            [["running",       "running"],
+             ["shutting-down", "shutting-down"],
+             ["shutting-down", "shutting-down"],
+             ["shutting-down", "terminated"],
+             ["terminated"]])
+
         self.ec2.describe_instances("i-amkillable", "i-amkillabletoo")
-        self.mocker.result(succeed([
-            self.get_instance("i-amkillable"),
-            self.get_instance("i-amkillabletoo")]))
+        self.mocker.call(system_state.get_round)
+        self.mocker.count(4)
+
+        self.ec2.describe_instances("i-amkillable")
+        self.mocker.call(system_state.get_round)
+
         self.ec2.terminate_instances("i-amkillable", "i-amkillabletoo")
         self.mocker.result(succeed([
             ("i-amkillable", "running", "shutting-down"),
             ("i-amkillabletoo", "running", "shutting-down")]))
+
+        deleted_groups = Observed()
+        self.ec2.delete_security_group("ensemble-moon-0")
+        self.mocker.call(deleted_groups.add)
+
+        self.ec2.delete_security_group("ensemble-moon-2")
+        self.mocker.call(deleted_groups.add)
+
         self.mocker.replay()
 
         provider = self.get_provider()
-        d = provider.shutdown_machines([
-            EC2ProviderMachine("i-amkillable"),
-            EC2ProviderMachine("i-amkillabletoo")])
-
-        def verify(result):
-            (machine_1, machine_2) = result
-            self.assertTrue(isinstance(machine_1, EC2ProviderMachine))
-            self.assertEquals(machine_1.instance_id, "i-amkillable")
-            self.assertTrue(isinstance(machine_2, EC2ProviderMachine))
-            self.assertEquals(machine_2.instance_id, "i-amkillabletoo")
-        d.addCallback(verify)
-        return d
+        machine_1, machine_2 = yield provider.shutdown_machines([
+                EC2ProviderMachine("i-amkillable"),
+                EC2ProviderMachine("i-amkillabletoo")])
+        self.assertTrue(isinstance(machine_1, EC2ProviderMachine))
+        self.assertEquals(machine_1.instance_id, "i-amkillable")
+        self.assertTrue(isinstance(machine_2, EC2ProviderMachine))
+        self.assertEquals(machine_2.instance_id, "i-amkillabletoo")
+        self.assertEquals(
+            deleted_groups.items,
+            set(["ensemble-moon-0", "ensemble-moon-2"]))
 
 
 class EC2DestroyTest(EC2TestMixin, TestCase):
 
+    @inlineCallbacks
     def test_destroy_environment(self):
         """
         The destroy_environment operation terminates all running and pending
-        instances associated to the C{MachineProvider} instance.
+        instances associated to the `MachineProvider` instance.
         """
         self.s3.put_object("moon", "provider-state", "{}\n")
         self.mocker.result(succeed(None))
         self.ec2.describe_instances()
-        self.mocker.result(succeed([
-            self.get_instance("i-canbekilled"),
-            self.get_instance("i-amdead", state="terminated"),
+        instances = [
+            self.get_instance("i-canbekilled", machine_id=0),
+            self.get_instance("i-amdead", machine_id=1, state="terminated"),
             self.get_instance("i-dontbelong", groups=["unknown"]),
-            self.get_instance("i-canbekilledtoo", state="pending")]))
+            self.get_instance(
+                "i-canbekilledtoo", machine_id=2, state="pending")]
+        self.mocker.result(succeed(instances))
         self.ec2.describe_instances("i-canbekilled", "i-canbekilledtoo")
         self.mocker.result(succeed([
-            self.get_instance("i-canbekilled"),
-            self.get_instance("i-canbekilledtoo", state="pending")]))
+                    self.get_instance("i-canbekilled"),
+                    self.get_instance("i-canbekilledtoo", state="pending")]))
         self.ec2.terminate_instances("i-canbekilled", "i-canbekilledtoo")
         self.mocker.result(succeed([
-            ("i-canbekilled", "running", "shutting-down"),
-            ("i-canbekilledtoo", "pending", "shutting-down")]))
+                    ("i-canbekilled", "running", "shutting-down"),
+                    ("i-canbekilledtoo", "pending", "shutting-down")]))
+
+        self.ec2.describe_instances("i-canbekilled", "i-canbekilledtoo")
+        states = [
+            self.get_instance(
+                "i-canbekilled", state="terminated", machine_id=0),
+            self.get_instance(
+                "i-canbekilledtoo", state="terminated", machine_id=2)]
+        self.mocker.result(succeed(states))
+
+        self.ec2.delete_security_group(MATCH_GROUP)
+        deleted_groups = Observed()
+        self.mocker.call(deleted_groups.add)
+        self.mocker.count(3)
+
         self.mocker.replay()
 
         provider = self.get_provider()
-        d = provider.destroy_environment()
-
-        def verify(result):
-            (machine_1, machine_2) = result
-            self.assertTrue(isinstance(machine_1, EC2ProviderMachine))
-            self.assertEquals(machine_1.instance_id, "i-canbekilled")
-            self.assertTrue(isinstance(machine_2, EC2ProviderMachine))
-            self.assertEquals(machine_2.instance_id, "i-canbekilledtoo")
-        d.addCallback(verify)
-        return d
-
+        machine_1, machine_2 = yield provider.destroy_environment()
+        self.assertTrue(isinstance(machine_1, EC2ProviderMachine))
+        self.assertEquals(machine_1.instance_id, "i-canbekilled")
+        self.assertTrue(isinstance(machine_2, EC2ProviderMachine))
+        self.assertEquals(machine_2.instance_id, "i-canbekilledtoo")
+        self.assertEquals(
+            deleted_groups.items,
+            set(["ensemble-moon-0", "ensemble-moon-2", "ensemble-moon"]))
+
+    @inlineCallbacks
     def test_s3_failure(self):
         """Failing to store empty state should not stop us killing machines"""
         self.s3.put_object("moon", "provider-state", "{}\n")
         self.mocker.result(fail(SomeError()))
         self.ec2.describe_instances()
         self.mocker.result(succeed([self.get_instance("i-canbekilled")]))
+        system_state = MockInstanceState(
+            self, ["i-canbekilled"], [0],
+            [["running"],
+             ["shutting-down"],
+             ["shutting-down"],
+             ["shutting-down"],
+             ["terminated"]])
         self.ec2.describe_instances("i-canbekilled")
-        self.mocker.result(succeed([self.get_instance("i-canbekilled")]))
+        self.mocker.call(system_state.get_round)
+        self.mocker.count(5)
         self.ec2.terminate_instances("i-canbekilled")
         self.mocker.result(succeed([
             ("i-canbekilled", "running", "shutting-down")]))
+        self.ec2.delete_security_group("ensemble-moon-0")
+        self.mocker.result(succeed(True))
+        self.ec2.delete_security_group("ensemble-moon")
+        self.mocker.result(succeed(True))
         self.mocker.replay()
 
         provider = self.get_provider()
-        d = provider.destroy_environment()
-
-        def verify(result):
-            (machine,) = result
-            self.assertTrue(isinstance(machine, EC2ProviderMachine))
-            self.assertEquals(machine.instance_id, "i-canbekilled")
-        d.addCallback(verify)
-        return d
-
+        machine, = yield provider.destroy_environment()
+        self.assertTrue(isinstance(machine, EC2ProviderMachine))
+        self.assertEquals(machine.instance_id, "i-canbekilled")
+
+    @inlineCallbacks
     def test_shutdown_no_instances(self):
         """
         If there are no instances to shutdown, running the destroy_environment
@@ -197,12 +253,14 @@
         self.mocker.result(succeed(None))
         self.ec2.describe_instances()
         self.mocker.result(succeed([]))
+        self.ec2.delete_security_group("ensemble-moon")
+        self.mocker.result(fail(
+                self.get_ec2_error(
+                    "ensemble-moon",
+                    format="The security group %r does not exist"
+                    )))
         self.mocker.replay()
 
         provider = self.get_provider()
-        d = provider.destroy_environment()
-
-        def verify(result):
-            self.assertEquals(result, [])
-        d.addCallback(verify)
-        return d
+        result = yield provider.destroy_environment()
+        self.assertEquals(result, [])