AMBARI-749. Complete Java side implementation of bootstrapping agent hosts. (mahadev)

git-svn-id: https://svn.apache.org/repos/asf/incubator/ambari/branches/AMBARI-666@1387460 13f79535-47bb-0310-9956-ffa450edef68

AMBARI-666-CHANGES.txt

@@ -12,6 +12,9 @@ AMBARI-666 branch (unreleased changes)
 
   NEW FEATURES
 
+  AMBARI-749. Complete Java side implementation of bootstrapping agent hosts.
+  (mahadev)
+
   AMBARI-757. Implement Installer Step 4 (Select Services). (yusaku)
 
   AMBARI-751. Re-structure servicecomponenthost fsm layout. (hitesh)

ambari-project/pom.xml

@@ -84,6 +84,11 @@
   </profiles>
   <dependencyManagement>
     <dependencies>
+      <dependency>
+        <groupId>commons-io</groupId>
+        <artifactId>commons-io</artifactId>
+        <version>2.1</version>
+      </dependency>
       <dependency>
         <groupId>com.google.inject</groupId>
         <artifactId>guice</artifactId>

ambari-server/pom.xml

@@ -56,6 +56,10 @@
   <profiles>
   </profiles>
   <dependencies>
+    <dependency>
+      <groupId>commons-io</groupId>
+      <artifactId>commons-io</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.google.inject</groupId>
       <artifactId>guice</artifactId>

ambari-server/src/main/java/org/apache/ambari/server/bootstrap/BSHostStatusCollector.java

@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.bootstrap;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * Runnable class that gets the hoststatus output by looking at the files
+ * in a certain directory. Only meant to be useful for bootstrap as of now.
+ */
+class BSHostStatusCollector {
+  private File requestIdDir;
+  private List<BSHostStatus> hostStatus;
+  static String logFileFilter = ".log";
+  static String doneFileFilter = ".done";
+  private static Log LOG = LogFactory.getLog(BSHostStatusCollector.class);
+  
+  private List<String> hosts;
+
+  public BSHostStatusCollector(File requestIdDir, List<String> hosts) {
+    this.requestIdDir = requestIdDir;
+    this.hosts = hosts;
+  }
+  
+  public List<BSHostStatus> getHostStatus() {
+    return hostStatus;
+  }
+  
+  public void run() {
+    LOG.info("Request directory " + requestIdDir);
+    hostStatus = new ArrayList<BSHostStatus>();
+    if (hosts == null) {
+      return;
+    }
+    File done;
+    File log;
+    LOG.info("HostList for polling on " + hosts);
+    for (String host: hosts) {
+      /* Read through the files and gather output */
+      BSHostStatus status = new BSHostStatus();
+      status.setHostName(host);
+      done = new File(requestIdDir, host + doneFileFilter);
+      log = new File(requestIdDir, host + logFileFilter);
+      if (!done.exists()) 
+        status.setStatus("RUNNING");
+      else 
+        status.setStatus("DONE");
+      if (!log.exists()) {
+        status.setLog("");
+      } else {
+        String logString = "";
+        try {
+          logString = FileUtils.readFileToString(log);
+        } catch (IOException e) {
+          LOG.info("Error reading log file " + log);
+        }
+        status.setLog(logString);
+      }
+      hostStatus.add(status);
+    }
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/bootstrap/BootStrapImpl.java

@@ -20,13 +20,20 @@ package org.apache.ambari.server.bootstrap;
 
 import java.io.File;
 import java.io.IOException;
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
+import java.io.StringWriter;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.ambari.server.bootstrap.BSResponse.BSRunStat;
+import org.apache.ambari.server.bootstrap.BootStrapStatus.BSStat;
 import org.apache.ambari.server.configuration.Configuration;
-import org.mortbay.log.Log;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
@@ -37,26 +44,23 @@ public class BootStrapImpl {
   private File bootStrapDir;
   private String bootScript;
   private BSRunner bsRunner;
- 
+  long timeout;
+
+  private static Log LOG = LogFactory.getLog(BootStrapImpl.class);
+
   /* Monotonically increasing requestid for the bootstrap api to query on */
-  AtomicLong requestId = new AtomicLong();
-  private static final int MAX_ENTRIES = 100;
+  int requestId = 0;
   private FifoLinkedHashMap<Long, BootStrapStatus> bsStatus;
 
-  /**
-   * Only Store the Last 100 BootStrap Status'es
-   *
-   */
-  @SuppressWarnings("serial")
-  public static class FifoLinkedHashMap<K, V> extends 
-  LinkedHashMap<K, V> {
-    protected boolean removeEldestEntry(Map.Entry<K, 
-        V> eldest) {
-      return size() > MAX_ENTRIES;
-    }
 
+  @Inject
+  public BootStrapImpl(Configuration conf) {
+    this.conf = conf;
+    this.bootStrapDir = conf.getBootStrapDir();
+    this.bootScript = conf.getBootStrapScript();   
+    this.bsStatus = new FifoLinkedHashMap<Long, BootStrapStatus>();
   }
-  
+
   /**
    * Return {@link BootStrapStatus} for a given responseId.
    * @param requestId the responseId for which the status needs to be returned.
@@ -69,27 +73,169 @@ public class BootStrapImpl {
     }
     return bsStatus.get(requestId);
   }
-  
+
+  /**
+   * update status of a request. Mostly called by the status collector thread.
+   * @param requestId the request id.
+   * @param status the status of the update.
+   */
+  private synchronized void updateStatus(long requestId, BootStrapStatus status) {
+    bsStatus.put(requestId, status);
+  }
+
+
+  /**
+   * Run the bs script to ssh to a list of hosts 
+   * with a ssh key.
+   */
   class BSRunner extends Thread {
     private  boolean finished = false;
     private SshHostInfo sshHostInfo;
-    private String bootDir;
-    private String bsString;
+    private File bootDir;
     private String bsScript;
+    private File requestIdDir;
+    private File sshKeyFile;
     int requestId;
+
     public BSRunner(SshHostInfo sshHostInfo, String bootDir, String bsScript,
-        int requestId)
+        int requestId, long timeout)
     {
       this.requestId = requestId;
       this.sshHostInfo = sshHostInfo;
-      this.bsString = bsString;
       this.bsScript = bsScript;
+      this.bootDir = new File(bootDir);
+      this.requestIdDir = new File(bootDir, Integer.toString(requestId));
+      this.sshKeyFile = new File(this.requestIdDir, "sshKey");
+      BootStrapStatus status = new BootStrapStatus();
+      status.setLog("RUNNING");
+      status.setStatus(BSStat.RUNNING);
+      BootStrapImpl.this.updateStatus(requestId, status);
     }
-    String[] command = new String[3];
-    
+
+    /**
+     * Update the gathered data from reading output
+     *
+     */
+    private class BSStatusCollector implements Runnable {
+      @Override
+      public void run() {
+        BSHostStatusCollector collector = new BSHostStatusCollector(requestIdDir, 
+            sshHostInfo.getHosts());
+        collector.run();
+        List<BSHostStatus> hostStatus = collector.getHostStatus();
+        BootStrapStatus status = new BootStrapStatus();
+        status.setHostsStatus(hostStatus);
+        status.setLog("");
+        status.setStatus(BSStat.RUNNING);
+        BootStrapImpl.this.updateStatus(requestId, status);
+      }
+    }
+
+    private String createHostString(List<String> list) {
+      String ret = "";
+      if (list == null) {
+        return ret;
+      }
+      for (String host: list) {
+        ret = ret + host + ",";
+      }
+      return ret;
+    }
+
+    /** Create request id dir for each bootstrap call **/
+    private void createRunDir() throws IOException {
+      if (!bootDir.exists()) {
+        // create the bootdir directory.
+        if (! bootDir.mkdirs()) {
+          throw new IOException("Cannot create " + bootDir);
+        }
+      }
+      /* create the request id directory */
+      if (requestIdDir.exists()) {
+        /* delete the directory and make sure we start back */
+        FileUtils.deleteDirectory(requestIdDir);
+      }
+      /* create the directory for the run dir */
+      if (! requestIdDir.mkdirs()) {
+        throw new IOException("Cannot create " + requestIdDir);
+      }
+    }
+
+    private void writeSshKeyFile(String data) throws IOException {
+      FileUtils.writeStringToFile(sshKeyFile, data);
+    }
+
+    public synchronized void finished() {
+      this.finished = true;
+    }
+
     @Override
-    public synchronized void run() {
-      
+    public void run() {
+      String hostString = createHostString(sshHostInfo.getHosts());
+      String commands[] = new String[4];
+      BSStat stat = BSStat.RUNNING;
+      String scriptlog = "";
+      try {
+        createRunDir();
+        writeSshKeyFile(sshHostInfo.getSshKey());
+        /* Running command: 
+         * script hostlist bsdir sshkeyfile
+         */
+        commands[0] = this.bsScript;
+        commands[1] = hostString;
+        commands[2] = this.requestIdDir.toString();
+        commands[3] = this.sshKeyFile.toString();
+        LOG.info("Host= " + hostString + " bs=" + this.bsScript + " requestDir=" +
+            requestIdDir + " keyfile=" + this.sshKeyFile);
+        Process process = Runtime.getRuntime().exec(commands);
+        /** Startup a scheduled executor service to look through the logs 
+         */
+        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+        BSStatusCollector statusCollector = new BSStatusCollector();
+        ScheduledFuture<?> handle = scheduler.scheduleWithFixedDelay(statusCollector, 
+            0, 10, TimeUnit.SECONDS);
+        LOG.info("Kicking off the scheduler for polling on logs in " + 
+            this.requestIdDir);
+        try {
+          int exitCode = process.waitFor();
+          StringWriter writer_1 = new StringWriter();
+          IOUtils.copy(process.getInputStream(), writer_1);
+          String outMesg = writer_1.toString();
+          if (outMesg == null)  outMesg = "";
+          StringWriter writer_2 = new StringWriter();
+          IOUtils.copy(process.getErrorStream(), writer_2);
+          String errMesg = writer_2.toString();
+          if (errMesg == null)  errMesg = "";
+          scriptlog = outMesg + "\n" + errMesg;
+          if (exitCode != 0) {
+            stat = BSStat.ERROR;
+          } else {
+            stat = BSStat.SUCCESS;
+          }
+        } catch (InterruptedException e) {
+          throw new IOException(e);
+        } finally {
+          handle.cancel(true);
+          /* schedule a last update */
+          scheduler.schedule(new BSStatusCollector(), 0, TimeUnit.SECONDS);
+          scheduler.shutdownNow();
+          try {
+            scheduler.awaitTermination(10, TimeUnit.SECONDS);
+          } catch (InterruptedException e) {
+            LOG.info("Interruped while waiting for scheduler");
+          }
+          process.destroy();
+        }
+      } catch(IOException io) {
+        LOG.info("Error executing bootstrap ", io);
+      } finally {
+        /* get the bstatus */
+        BootStrapStatus tmpStatus = getStatus(requestId);
+        tmpStatus.setLog(scriptlog);
+        tmpStatus.setStatus(stat);
+        updateStatus(requestId, tmpStatus);
+        finished();
+      }
     }
 
     public synchronized boolean isRunning() {
@@ -97,34 +243,31 @@ public class BootStrapImpl {
     }
   }
 
-  @Inject
-  public BootStrapImpl(Configuration conf) {
-    this.conf = conf;
-    this.bootStrapDir = conf.getBootStrapDir();
-    this.bootScript = conf.getBootStrapScript();   
-    this.bsStatus = new FifoLinkedHashMap<Long, BootStrapStatus>();
-  }
-
   public synchronized void init() throws IOException {
-    boolean mkdirs = bootStrapDir.mkdirs();
-    if (!mkdirs) throw new IOException("Unable to make directory for " +
-        "bootstrap " + bootStrapDir);
-
+    if (!bootStrapDir.exists()) {
+      boolean mkdirs = bootStrapDir.mkdirs();
+      if (!mkdirs) throw new IOException("Unable to make directory for " +
+          "bootstrap " + bootStrapDir);
+    }
   }
 
   public  synchronized BSResponse runBootStrap(SshHostInfo info) {
     BSResponse response = new BSResponse();
     /* Run some checks for ssh host */
-    Log.info("BootStrapping hosts " + info.hostListAsString());
+    LOG.info("BootStrapping hosts " + info.hostListAsString());
     if (bsRunner != null) {
       response.setLog("BootStrap in Progress: Cannot Run more than one.");
       response.setStatus(BSRunStat.ERROR);
       return response;
     } 
-    
-    
+    requestId++;
+
+    bsRunner = new BSRunner(info, bootStrapDir.toString(),
+        bootScript, requestId, 0L);
+    bsRunner.start();
     response.setStatus(BSRunStat.OK);
     response.setLog("Running Bootstrap now.");
+    response.setRequestId(requestId);
     return response;
   }
 

ambari-server/src/main/java/org/apache/ambari/server/bootstrap/FifoLinkedHashMap.java

@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.bootstrap;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * Only Store the most recent 100 Key Value Pairs.
+ *
+ */
+@SuppressWarnings("serial")
+public class FifoLinkedHashMap<K, V> extends 
+LinkedHashMap<K, V> {
+  public static final int MAX_ENTRIES = 100;
+  protected boolean removeEldestEntry(Map.Entry<K, 
+      V> eldest) {
+    return size() > MAX_ENTRIES;
+  }
+
+}

ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java

@@ -34,8 +34,8 @@ import org.apache.commons.logging.LogFactory;
 public class Configuration {
   private static final String AMBARI_CONF_VAR = "AMBARI_CONF_DIR";
   private static final String CONFIG_FILE = "ambari.properties";
-  private static final String BOOTSTRAP_DIR = "bootstrap.dir";
-  private static final String BOOTSTRAP_SCRIPT = "bootstrap.script";
+  public static final String BOOTSTRAP_DIR = "bootstrap.dir";
+  public static final String BOOTSTRAP_SCRIPT = "bootstrap.script";
 
   private static final Log LOG = LogFactory.getLog(Configuration.class);
 

ambari-server/src/test/java/org/apache/ambari/server/bootstrap/BootStrapTest.java

@@ -18,22 +18,104 @@
 
 package org.apache.ambari.server.bootstrap;
 
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Properties;
 
+import junit.framework.Assert;
 import junit.framework.TestCase;
 
+import org.apache.ambari.server.bootstrap.BootStrapStatus.BSStat;
 import org.apache.ambari.server.configuration.Configuration;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.After;
+import org.junit.Before;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+
 /**
  * Test BootStrap Implementation.
  */
 public class BootStrapTest extends TestCase {
+  private static Log LOG = LogFactory.getLog(BootStrapTest.class);
+  public TemporaryFolder temp = new TemporaryFolder();
+  
+  @Before
+  public void setUp() throws IOException {
+    temp.create();
+  }
+  
+  @After
+  public void tearDown() throws IOException {
+    temp.delete();
+  }
   
   @Test
   public void testRun() throws Exception {
     Properties properties = new Properties();
+    String bootdir =  temp.newFolder("bootdir").toString();
+    LOG.info("Bootdir is " + bootdir);
+    properties.setProperty(Configuration.BOOTSTRAP_DIR, 
+       bootdir);
+    properties.setProperty(Configuration.BOOTSTRAP_SCRIPT, "echo");
     Configuration conf = new Configuration(properties);
     BootStrapImpl impl = new BootStrapImpl(conf);
+    impl.init();
+    SshHostInfo info = new SshHostInfo();
+    info.setSshKey("xyz");
+    ArrayList<String> hosts = new ArrayList<String>();
+    hosts.add("host1");
+    hosts.add("host2");
+    info.setHosts(hosts);
+    BSResponse response = impl.runBootStrap(info);
+    LOG.info("Response id from bootstrap " + response.getRequestId());
+    /* do a query */
+    BootStrapStatus status = impl.getStatus(response.getRequestId());
+    LOG.info("Status " + status.getStatus());
+    int num = 0;
+    while ((status.getStatus() != BSStat.SUCCESS) && (num < 10000)) {
+        status = impl.getStatus(response.getRequestId());
+        Thread.sleep(100);
+        num++;
+    }
+    LOG.info("Status: log " + status.getLog() + " status=" + status.getStatus() 
+        );
+    /* Note its an echo command so it should echo host1,host2 */
+    Assert.assertTrue(status.getLog().contains("host1,host2"));
+    Assert.assertEquals(BSStat.SUCCESS, status.getStatus());
+  }
+  
+  
+  @Test
+  public void testPolling() throws Exception {
+    File tmpFolder = temp.newFolder("bootstrap");
+    /* create log and done files */
+    FileUtils.writeStringToFile(new File(tmpFolder, "host1.done"), "output");
+    FileUtils.writeStringToFile(new File(tmpFolder, "host1.log"), "err_log_1");
+    FileUtils.writeStringToFile(new File(tmpFolder, "host2.done"), "output");
+    FileUtils.writeStringToFile(new File(tmpFolder, "host2.log"), "err_log_2");
+
+    List<String> listHosts = new ArrayList<String>();
+    listHosts.add("host1");
+    listHosts.add("host2");
+    BSHostStatusCollector collector = new BSHostStatusCollector(tmpFolder,
+        listHosts);
+    collector.run();
+    List<BSHostStatus> polledHostStatus = collector.getHostStatus();
+    Assert.assertTrue(polledHostStatus.size() == 2);
+    Assert.assertEquals(polledHostStatus.get(0).getHostName(), "host1");
+    Assert.assertEquals(polledHostStatus.get(0).getLog(), "err_log_1");
+    Assert.assertEquals(polledHostStatus.get(0).getStatus(), "DONE");
+    Assert.assertEquals(polledHostStatus.get(1).getHostName(), "host2");
+    Assert.assertEquals(polledHostStatus.get(1).getLog(), "err_log_2");
+    Assert.assertEquals(polledHostStatus.get(1).getStatus(), "DONE");
+    
+
   }
   
 }