WHIRR-398. Implement the execution of scripts on DestroyClusterAction…

… (David Alves via asavu)

git-svn-id: https://svn.apache.org/repos/asf/whirr/trunk@1198689 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -56,6 +56,9 @@ Trunk (unreleased changes)
 
     WHIRR-423. Refactor StartupProcess.cleanupFailedNodes (Adrian Cole via asavu)
 
+    WHIRR-398. Implement the execution of scripts on DestroyClusterAction 
+    (David Alves via asavu)
+
   BUG FIXES
 
     WHIRR-377. Fix broken CLI logging config. (asavu via tomwhite)

core/src/main/java/org/apache/whirr/ClusterController.java

@@ -116,15 +116,22 @@ public Cluster launchCluster(ClusterSpec clusterSpec)
 
   /**
    * Stop the cluster and destroy all resources associated with it.
-   *
-   * @throws IOException if there is a problem while stopping the cluster. The
-   * cluster may or may not have been stopped.
-   * @throws InterruptedException if the thread is interrupted.
+   * 
+   * @throws IOException
+   *           if there is a problem while stopping the cluster. The cluster may
+   *           or may not have been stopped.
+   * @throws InterruptedException
+   *           if the thread is interrupted.
    */
   public void destroyCluster(ClusterSpec clusterSpec) throws IOException,
       InterruptedException {
-    DestroyClusterAction destroyer = new DestroyClusterAction(getCompute());
-    destroyer.execute(clusterSpec, null);
+
+    ClusterStateStore store = getClusterStateStore(clusterSpec);
+    Cluster cluster = store.load();
+
+    DestroyClusterAction destroyer = new DestroyClusterAction(getCompute(),
+        HandlerMapFactory.create());
+    destroyer.execute(clusterSpec, cluster);
 
     getClusterStateStore(clusterSpec).destroy();
   }

core/src/main/java/org/apache/whirr/actions/DestroyClusterAction.java

@@ -18,32 +18,57 @@
 
 package org.apache.whirr.actions;
 
+import static org.apache.whirr.RolePredicates.onlyRolesIn;
+import static org.jclouds.compute.options.RunScriptOptions.Builder.overrideCredentialsWith;
 import static org.jclouds.compute.predicates.NodePredicates.inGroup;
 
 import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import javax.annotation.Nullable;
 
 import org.apache.whirr.Cluster;
-import org.apache.whirr.ClusterAction;
+import org.apache.whirr.Cluster.Instance;
 import org.apache.whirr.ClusterSpec;
+import org.apache.whirr.InstanceTemplate;
+import org.apache.whirr.service.ClusterActionEvent;
 import org.apache.whirr.service.ClusterActionHandler;
+import org.apache.whirr.service.jclouds.StatementBuilder;
 import org.jclouds.compute.ComputeService;
 import org.jclouds.compute.ComputeServiceContext;
+import org.jclouds.compute.domain.ExecResponse;
+import org.jclouds.domain.Credentials;
+import org.jclouds.scriptbuilder.domain.OsFamily;
+import org.jclouds.scriptbuilder.domain.Statement;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Sets;
 
 /**
- * A {@link ClusterAction} for tearing down a running cluster and freeing up
- * all its resources.
+ * A {@link ClusterAction} for tearing down a running cluster and freeing up all
+ * its resources.
  */
-public class DestroyClusterAction extends ClusterAction {
+public class DestroyClusterAction extends ScriptBasedClusterAction {
+
+  private static final Logger LOG = LoggerFactory
+      .getLogger(DestroyClusterAction.class);
 
-  private static final Logger LOG =
-    LoggerFactory.getLogger(DestroyClusterAction.class);
-
-  public DestroyClusterAction(Function<ClusterSpec, ComputeServiceContext> getCompute) {
-    super(getCompute);
+  public DestroyClusterAction(
+      final Function<ClusterSpec, ComputeServiceContext> getCompute,
+      final Map<String, ClusterActionHandler> handlerMap) {
+    super(getCompute, handlerMap);
   }
 
   @Override
@@ -52,13 +77,95 @@ protected String getAction() {
   }
 
   @Override
-  public Cluster execute(ClusterSpec clusterSpec, Cluster cluster)
+  protected void doAction(Map<InstanceTemplate, ClusterActionEvent> eventMap)
       throws IOException, InterruptedException {
+
+    final ExecutorService executorService = Executors.newCachedThreadPool();
+    final Collection<Future<ExecResponse>> futures = Sets.newHashSet();
+
+    ClusterSpec clusterSpec = eventMap.values().iterator().next()
+        .getClusterSpec();
+
+    for (Entry<InstanceTemplate, ClusterActionEvent> entry : eventMap
+        .entrySet()) {
+
+      Cluster cluster = entry.getValue().getCluster();
+
+      StatementBuilder statementBuilder = entry.getValue()
+          .getStatementBuilder();
+
+      ComputeServiceContext computeServiceContext = getCompute().apply(
+          clusterSpec);
+      final ComputeService computeService = computeServiceContext
+          .getComputeService();
+
+      final Credentials credentials = new Credentials(
+          clusterSpec.getClusterUser(), clusterSpec.getPrivateKey());
+
+      Set<Instance> instances = cluster.getInstancesMatching(onlyRolesIn(entry
+          .getKey().getRoles()));
+
+      String instanceIds = Joiner.on(", ").join(
+          Iterables.transform(instances, new Function<Instance, String>() {
+            @Override
+            public String apply(@Nullable Instance instance) {
+              return instance == null ? "<null>" : instance.getId();
+            }
+          }));
+
+      LOG.info("Starting to run destroy scripts on cluster " + "instances: {}",
+          instanceIds);
+
+      for (final Instance instance : instances) {
+        final Statement statement = statementBuilder.build(clusterSpec,
+            instance);
+
+        futures.add(executorService.submit(new Callable<ExecResponse>() {
+          @Override
+          public ExecResponse call() {
+
+            LOG.info("Running destroy script on: {}", instance.getId());
+            if (LOG.isDebugEnabled()) {
+              LOG.debug("Destroy script for {}:\n{}", instance.getId(),
+                  statement.render(OsFamily.UNIX));
+            }
+
+            try {
+              return computeService.runScriptOnNode(
+                  instance.getId(),
+                  statement,
+                  overrideCredentialsWith(credentials)
+                      .runAsRoot(true)
+                      .nameTask(
+                          "destroy-" + Joiner.on('_').join(instance.getRoles())));
+
+            } finally {
+              LOG.info("Destroy script run completed on: {}", instance.getId());
+            }
+          }
+        }));
+      }
+    }
+
+    for (Future<ExecResponse> future : futures) {
+      try {
+        ExecResponse execResponse = future.get();
+        if (execResponse.getExitCode() != 0) {
+          LOG.error("Error running script: {}\n{}", execResponse.getError(),
+              execResponse.getOutput());
+        }
+      } catch (ExecutionException e) {
+        throw new IOException(e.getCause());
+      }
+    }
+
+    LOG.info("Finished running destroy scripts on all cluster instances.");
+
     LOG.info("Destroying " + clusterSpec.getClusterName() + " cluster");
-    ComputeService computeService = getCompute().apply(clusterSpec).getComputeService();
+    ComputeService computeService = getCompute().apply(clusterSpec)
+        .getComputeService();
     computeService.destroyNodesMatching(inGroup(clusterSpec.getClusterName()));
     LOG.info("Cluster {} destroyed", clusterSpec.getClusterName());
-    return null;
   }
 
 }