cloudmgr fix (still experimental)

VonRosenchild · Jul 20, 2011 · b19e1bc · b19e1bc
2 parents 781e2c0 + 9f7550a
commit b19e1bc
Show file tree

Hide file tree

Showing 45 changed files with 363 additions and 152 deletions.
diff --git a/build-tools/pom.xml b/build-tools/pom.xml
@@ -7,12 +7,12 @@
  <parent>
  <artifactId>stratosphere</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>build-tools</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  <name>build-tools</name>
  <packaging>jar</packaging>
 </project>
diff --git a/nephele/nephele-clustermanager/pom.xml b/nephele/nephele-clustermanager/pom.xml
@@ -3,15 +3,15 @@
  <parent>
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <modelVersion>4.0.0</modelVersion>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-clustermanager</artifactId>
  <name>nephele-clustermanager</name>
- <version>0.1</version>
+ <version>0.1.1</version>
  <url>https://maven.apache.org</url>
 
  <dependencies>

diff --git a/nephele/nephele-common/pom.xml b/nephele/nephele-common/pom.xml
@@ -6,12 +6,12 @@
  <parent>
  <artifactId>nephele</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-common</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  <name>nephele-common</name>
 
  <packaging>jar</packaging>

diff --git a/nephele/nephele-common/src/main/java/eu/stratosphere/nephele/client/JobClient.java b/nephele/nephele-common/src/main/java/eu/stratosphere/nephele/client/JobClient.java
@@ -24,7 +24,6 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
-import eu.stratosphere.nephele.client.AbstractJobResult.ReturnCode;
 import eu.stratosphere.nephele.configuration.ConfigConstants;
 import eu.stratosphere.nephele.configuration.Configuration;
 import eu.stratosphere.nephele.event.job.AbstractEvent;
@@ -111,6 +110,8 @@ public void run() {
  // Terminate the running job if the configuration says so
  if (this.jobClient.getConfiguration().getBoolean(ConfigConstants.JOBCLIENT_SHUTDOWN_TERMINATEJOB_KEY,
  ConfigConstants.DEFAULT_JOBCLIENT_SHUTDOWN_TERMINATEJOB)) {
+ System.out.println(AbstractEvent.timestampToString(System.currentTimeMillis())
+ + ":\tJobClient is shutting down, canceling job...");
  this.jobClient.cancelJob();
  }
 
@@ -221,12 +222,7 @@ public JobSubmissionResult submitJob() throws IOException {
 
  synchronized (this.jobSubmitClient) {
 
- final JobSubmissionResult result = this.jobSubmitClient.submitJob(this.jobGraph);
- if (result.getReturnCode() == ReturnCode.SUCCESS) {
- // Make sure the job is properly terminated when the user shut's down the client
- Runtime.getRuntime().addShutdownHook(this.jobCleanUp);
- }
- return result;
+ return this.jobSubmitClient.submitJob(this.jobGraph);
  }
  }
 
@@ -239,8 +235,6 @@ public JobSubmissionResult submitJob() throws IOException {
  */
  public JobCancelResult cancelJob() throws IOException {
 
- Runtime.getRuntime().removeShutdownHook(this.jobCleanUp);
-
  synchronized (this.jobSubmitClient) {
  return this.jobSubmitClient.cancelJob(this.jobGraph.getJobID());
  }
@@ -277,10 +271,10 @@ public void submitJobAndWait() throws IOException, JobExecutionException {
  if (submissionResult.getReturnCode() == AbstractJobResult.ReturnCode.ERROR) {
  LOG.error("ERROR: " + submissionResult.getDescription());
  throw new JobExecutionException(submissionResult.getDescription(), false);
- } else {
- // Make sure the job is properly terminated when the user shut's down the client
- Runtime.getRuntime().addShutdownHook(this.jobCleanUp);
  }
+
+ // Make sure the job is properly terminated when the user shut's down the client
+ Runtime.getRuntime().addShutdownHook(this.jobCleanUp);
  }
 
  long sleep = 0;
@@ -334,7 +328,8 @@ public void submitJobAndWait() throws IOException, JobExecutionException {
  } else if (jobStatus == JobStatus.CANCELED || jobStatus == JobStatus.FAILED) {
  Runtime.getRuntime().removeShutdownHook(this.jobCleanUp);
  LOG.info(jobEvent.getOptionalMessage());
- throw new JobExecutionException(jobEvent.getOptionalMessage(), (jobStatus == JobStatus.CANCELED) ? true : false);
+ throw new JobExecutionException(jobEvent.getOptionalMessage(),
+ (jobStatus == JobStatus.CANCELED) ? true : false);
  }
  }
  }

diff --git a/nephele/nephele-common/src/main/java/eu/stratosphere/nephele/event/job/AbstractEvent.java b/nephele/nephele-common/src/main/java/eu/stratosphere/nephele/event/job/AbstractEvent.java
@@ -96,7 +96,7 @@ public long getTimestamp() {
  * the timestamp in milliseconds since the beginning of "the epoch"
  * @return the string unified representation of the timestamp
  */
- protected static String timestampToString(long timestamp) {
+ public static String timestampToString(long timestamp) {
 
  return dateFormatter.format(new Date(timestamp));
 

diff --git a/nephele/nephele-compression-bzip2/pom.xml b/nephele/nephele-compression-bzip2/pom.xml
@@ -7,12 +7,12 @@
  <parent>
  <artifactId>nephele</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-compression-bzip2</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  <name>nephele-compression-bzip2</name>
 
  <dependencies>

diff --git a/nephele/nephele-compression-lzma/pom.xml b/nephele/nephele-compression-lzma/pom.xml
@@ -7,12 +7,12 @@
  <parent>
  <artifactId>nephele</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-compression-lzma</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  <name>nephele-compression-lzma</name>
 
  <dependencies>

diff --git a/nephele/nephele-compression-snappy/pom.xml b/nephele/nephele-compression-snappy/pom.xml
@@ -7,12 +7,12 @@
  <parent>
  <artifactId>nephele</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-compression-snappy</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  <name>nephele-compression-snappy</name>
 
  <dependencies>

diff --git a/nephele/nephele-compression-zlib/pom.xml b/nephele/nephele-compression-zlib/pom.xml
@@ -7,12 +7,12 @@
  <parent>
  <artifactId>nephele</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-compression-zlib</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  <name>nephele-compression-zlib</name>
 
  <dependencies>

diff --git a/nephele/nephele-ec2cloudmanager/pom.xml b/nephele/nephele-ec2cloudmanager/pom.xml
@@ -7,15 +7,15 @@
  <parent>
  <artifactId>nephele</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-ec2cloudmanager</artifactId>
  <name>nephele-ec2cloudmanager</name>
  <packaging>jar</packaging>
 
- <version>0.1</version>
+ <version>0.1.1</version>
 
  <dependencies>
  <dependency>

diff --git a/...le-ec2cloudmanager/src/main/java/eu/stratosphere/nephele/instance/cloud/CloudManager.java b/...le-ec2cloudmanager/src/main/java/eu/stratosphere/nephele/instance/cloud/CloudManager.java
@@ -543,8 +543,8 @@ public void requestInstance(JobID jobID, Configuration conf, Map<InstanceType, I
  * }
  */
 
- final String awsAccessId = "AKIAJYQJNI7QH227NDQA";
- final String awsSecretKey = "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";
+ final String awsAccessId = "HERE";
+ final String awsSecretKey = "HERE";
  final String owner = "nobody";
 
  final String sshKeyPair = conf.getString("job.cloud.sshkeypair", null);
@@ -640,7 +640,7 @@ private LinkedList<String> allocateCloudInstance(String awsAccessId, String awsS
  * return null;
  * }
  */
- final String imageID = "ami-ea5b6b9e";
+ final String imageID = "ami-a24272d6";
  final String jobManagerIPAddress = GlobalConfiguration.getString("jobmanager.rpc.address", null);
  if (jobManagerIPAddress == null) {
  LOG.error("JobManager IP address is not set (jobmanager.rpc.address)");
@@ -750,6 +750,30 @@ private LinkedList<CloudInstance> anyFloatingInstanceAvailable(String owner, Str
 
  // Check if instance entry is a floating instance
 
+
+ final Iterator<Map.Entry<InstanceConnectionInfo, FloatingInstance>> it = this.floatingInstances.entrySet().iterator();
+
+ while (it.hasNext()) {
+
+ Entry<InstanceConnectionInfo, FloatingInstance> e = it.next();
+ if (e.getKey().getAddress().equals(inetAddress)) {
+ LOG.info("Suitable floating instance found.");
+ final FloatingInstance floatingInstance = e.getValue();
+ it.remove();
+
+ this.floatingInstanceIDs.remove(floatingInstance.getInstanceID());
+
+ floatinginstances.add(convertIntoCloudInstance(t,
+ floatingInstance.getInstanceConnectionInfo(), owner));
+
+ // If we already have enough floating instances found: return!
+ if (floatinginstances.size() >= count) {
+ return floatinginstances;
+ }
+ }
+
+ }
+
  for (Entry<InstanceConnectionInfo, FloatingInstance> e : this.floatingInstances.entrySet()) {
  if (e.getKey().getAddress().equals(inetAddress)) {
  LOG.info("Suitable floating instance found.");

diff --git a/nephele/nephele-examples/pom.xml b/nephele/nephele-examples/pom.xml
@@ -7,12 +7,12 @@
  <parent>
  <artifactId>nephele</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-examples</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  <name>nephele-examples</name>
  <packaging>jar</packaging>
 

diff --git a/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/GrepTask.java b/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/GrepTask.java
@@ -35,6 +35,7 @@ public void invoke() throws Exception {
 
  StringRecord s = input.next();
  this.output.emit(s);
+ i++;
  }
 
  System.out.println("GREP: Emmited all " + i + " records");

diff --git a/nephele/nephele-hdfs/pom.xml b/nephele/nephele-hdfs/pom.xml
@@ -7,12 +7,12 @@
  <parent>
  <artifactId>nephele</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-hdfs</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  <name>nephele-hdfs</name>
  <packaging>jar</packaging>
 

diff --git a/nephele/nephele-management/pom.xml b/nephele/nephele-management/pom.xml
@@ -8,12 +8,12 @@
  <parent>
  <artifactId>nephele</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-management</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  <name>nephele-management</name>
  <packaging>jar</packaging>
 

diff --git a/nephele/nephele-profiling/pom.xml b/nephele/nephele-profiling/pom.xml
@@ -8,12 +8,12 @@
  <parent>
  <artifactId>nephele</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-profiling</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  <name>nephele-profiling</name>
  <packaging>jar</packaging>
 

diff --git a/nephele/nephele-queuescheduler/pom.xml b/nephele/nephele-queuescheduler/pom.xml
@@ -3,13 +3,13 @@
  <parent>
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
  <modelVersion>4.0.0</modelVersion>
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-queuescheduler</artifactId>
  <name>nephele-queuescheduler</name>
- <version>0.1</version>
+ <version>0.1.1</version>
  <url>https://maven.apache.org</url>
 
  <dependencies>

diff --git a/nephele/nephele-s3/pom.xml b/nephele/nephele-s3/pom.xml
@@ -7,12 +7,12 @@
  <parent>
  <artifactId>nephele</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-s3</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  <name>nephele-s3</name>
  <packaging>jar</packaging>
 

diff --git a/nephele/nephele-server/pom.xml b/nephele/nephele-server/pom.xml
@@ -6,12 +6,12 @@
  <parent>
  <artifactId>nephele</artifactId>
  <groupId>eu.stratosphere</groupId>
- <version>0.1</version>
+ <version>0.1.1</version>
  </parent>
 
  <groupId>eu.stratosphere</groupId>
  <artifactId>nephele-server</artifactId>
- <version>0.1</version>
+ <version>0.1.1</version>
  <name>nephele-server</name>
 
  <packaging>jar</packaging>