Added member ports summary in Timeline proposals

src/main/java/water/H2O.java

@@ -389,6 +389,7 @@ public static class OptArgs extends Arguments.Opt {
     public String keepice; // Do not delete ice on startup
     public String soft = null; // soft launch for demos
     public String random_udp_drop = null; // test only, randomly drop udp incoming
+    public String log_headers = null; // add machine name, PID and time to logs
   }
   public static boolean IS_SYSTEM_RUNNING = false;
 
@@ -404,6 +405,9 @@ public static void main( String[] args ) {
     arguments.extract(OPT_ARGS);
     ARGS = arguments.toStringArray();
 
+    if(OPT_ARGS.log_headers != null)
+      Log.initHeaders();
+
     startLocalNode(); // start the local node
     // Load up from disk and initialize the persistence layer
     initializePersistence();

src/main/java/water/Log.java

@@ -3,15 +3,14 @@
 import java.io.*;
 import java.lang.management.ManagementFactory;
 import java.text.SimpleDateFormat;
-import java.util.Date;
-import java.util.TimeZone;
+import java.util.*;
 
 public final class Log {
   // @formatter:off
   private static final ThreadLocal<SimpleDateFormat> _utcFormat = new ThreadLocal<SimpleDateFormat>() {
     @Override
     protected SimpleDateFormat initialValue() {
-      SimpleDateFormat format = new SimpleDateFormat("yyyyMMdd'-'HH:mm:ss.SSS");
+      SimpleDateFormat format = new SimpleDateFormat("yyMMdd'-'HH:mm:ss.SSS");
       format.setTimeZone(TimeZone.getTimeZone("UTC"));
       return format;
     }
@@ -20,17 +19,16 @@ protected SimpleDateFormat initialValue() {
   // @formatter:on
 
   static {
-    HOST_AND_PID = ""
-        + padRight(H2O.findInetAddressForSelf().getHostAddress() + ", ", 17)
-        + padRight(getPid() + ", ", 8);
+    HOST_AND_PID = "" + padRight(H2O.findInetAddressForSelf().getHostAddress() + ", ", 17) + padRight(getPid() + ", ", 8);
   }
 
   private static long getPid() {
     try {
       String n = ManagementFactory.getRuntimeMXBean().getName();
       int i = n.indexOf('@');
-      if( i == -1 ) return -1;
-      return Long.parseLong(n.substring(0,i));
+      if( i == -1 )
+        return -1;
+      return Long.parseLong(n.substring(0, i));
     } catch( Throwable t ) {
       return -1;
     }
@@ -52,10 +50,7 @@ public static void write(String s, Throwable t) {
       t.printStackTrace(printWriter);
       stack = result.toString();
     }
-    String header = _utcFormat.get().format(new Date()) + ", ";
-    header += HOST_AND_PID;
-    header += padRight(Thread.currentThread().getName() + ", ", 12);
-    System.out.println(header + (s != null ? s + " " + stack : stack));
+    System.out.println(s != null ? s + " " + stack : stack);
   }
 
   // Print to the original STDERR & die
@@ -73,4 +68,48 @@ static String padRight(String stringToPad, int size) {
 
     return strb.toString();
   }
+
+  static void initHeaders() {
+    System.setOut(new Wrapper(System.out));
+    System.setErr(new Wrapper(System.err));
+  }
+
+  public static void write(PrintStream stream, String s, boolean headers) {
+    if( stream instanceof Wrapper )
+      ((Wrapper) System.out).printlnParent(s);
+    else
+      stream.println(s);
+  }
+
+  private static final class Wrapper extends PrintStream {
+    Wrapper(PrintStream parent) {
+      super(parent);
+    }
+
+    static String h() {
+      String h = _utcFormat.get().format(new Date()) + ", ";
+      h += HOST_AND_PID;
+      h += padRight(Thread.currentThread().getName() + ", ", 26);
+      return h;
+    }
+
+    @Override
+    public PrintStream printf(String format, Object... args) {
+      return super.printf(h() + format, args);
+    }
+
+    @Override
+    public PrintStream printf(Locale l, String format, Object... args) {
+      return super.printf(l, h() + format, args);
+    }
+
+    @Override
+    public void println(String x) {
+      super.println(h() + x);
+    }
+
+    void printlnParent(String s) {
+      super.println(s);
+    }
+  }
 }

src/main/java/water/Paxos.java

@@ -205,7 +205,7 @@ synchronized static void doChangeAnnouncement( H2O cloud ) {
       _state._idLo = uuid.getLeastSignificantBits();
       _state._idHi = uuid.getMostSignificantBits();
       Paxos.print("send: Prepare "+proposal_num+" for leadership fight ",PROPOSED_MEMBERS);
-      UDPPaxosProposal.build_and_multicast(proposal_num);
+      UDPPaxosProposal.build_and_multicast(proposal_num, _state._members);
     } else {
       // Non-Leaders act as passive Accepters.  All Nodes should respond in a
       // timely fashion, including Leaders - if they fail the basic heartbeat

src/main/java/water/UDPPaxosProposal.java

@@ -8,18 +8,28 @@
  */
 
 public class UDPPaxosProposal extends UDP {
+  private static final int PRINT_PORTS_COUNT = 5;
+
   @Override AutoBuffer call(AutoBuffer ab) {
     if( ab._h2o._heartbeat != null ) Paxos.doProposal(ab.get8(), ab._h2o);
     return ab;
   }
 
-  static void build_and_multicast( final long proposal_num ) {
-    new AutoBuffer(H2O.SELF).putUdp(udp.paxos_proposal).put8(proposal_num).close();
+  static void build_and_multicast( final long proposal_num, H2ONode[] members ) {
+    AutoBuffer bb = new AutoBuffer(H2O.SELF).putUdp(udp.paxos_proposal).put8(proposal_num);
+    byte[] ports = new byte[PRINT_PORTS_COUNT];
+    for(int i = Math.min(members.length - 1, ports.length - 1); i >= 0; i--)
+      ports[i] = (byte) (members[i]._key.htm_port() % 100);
+    bb.putA1(ports, ports.length);
+    bb.close();
   }
 
   // Pretty-print bytes 1-15; byte 0 is the udp_type enum
   public String print16( AutoBuffer ab ) {
     ab.getPort();
-    return "Proposal# "+ab.get8();
+    String s = "Proposal# " + ab.get8();
+    for( int i = 0; i < PRINT_PORTS_COUNT; i++ )
+      s += ", " + ab.get1();
+    return s;
   }
 }

src/test/java/water/util/Sandbox.java

@@ -25,7 +25,8 @@ public static void main(String[] args) throws Exception {
 
     Desktop desktop = Desktop.getDesktop();
     // desktop.browse(new URI("http://localhost:54321/Jobs.html"));
-    desktop.browse(new URI("http://localhost:54321/Inspect.html?key=test.hex"));
+    // desktop.browse(new URI("http://localhost:54321/Inspect.html?key=test.hex"));
+    desktop.browse(new URI("http://localhost:54321/Timeline.html"));
 
     BufferedReader console = new BufferedReader(new InputStreamReader(System.in));
     console.readLine();

src/test/java/water/util/SeparateVM.java

@@ -1,12 +1,11 @@
 package water.util;
 
-import H2OInit.Boot;
-
 import java.io.*;
 import java.util.ArrayList;
 import java.util.Arrays;
 
 import water.Log;
+import H2OInit.Boot;
 
 /**
  * Executes code in a separate VM.
@@ -77,28 +76,27 @@ public void run() {
   }
 
   static void inheritIO(Process process, final String description, final boolean addLogHeader) {
-    final BufferedReader input = new BufferedReader(new InputStreamReader(process.getInputStream()));
+    forward(process, description, addLogHeader, process.getInputStream(), System.out);
+    forward(process, description, addLogHeader, process.getErrorStream(), System.err);
+  }
+
+  private static void forward(Process process, final String description, final boolean headers, //
+      InputStream in, final PrintStream out) {
+    final BufferedReader stream = new BufferedReader(new InputStreamReader(in));
     Thread thread = new Thread() {
       @Override
       public void run() {
-        if( description != null ) {
-          if( addLogHeader )
-            Log.write(description);
-          else
-            System.out.println(description);
-        }
+        if( description != null )
+          Log.write(out, description, headers);
 
         try {
           for( ;; ) {
-            String line = input.readLine();
+            String line = stream.readLine();
 
             if( line == null )
               break;
 
-            if( addLogHeader )
-              Log.write(line);
-            else
-              System.out.println(line);
+            Log.write(out, line, headers);
           }
         } catch( IOException e ) {
           // Ignore, process probably done

src/test/java/water/util/TestRunner.java

@@ -1,9 +1,13 @@
 package water.util;
 
-import hex.KMeansTest;
-
+import java.awt.Desktop;
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.net.URI;
 import java.util.ArrayList;
 
+import water.H2O;
+
 /**
  * Builds a cloud by duplicating current JVM settings in different processes or machines, and runs tests.
  */
@@ -17,15 +21,22 @@ public static void main(String[] args) throws Exception {
 
     for( int i = 0; i < nodes - 1; i++ ) {
       // sites.add(new SeparateCL());
-      sites.add(new SeparateVM("VM" + i, null));
+      sites.add(new SeparateVM("VM" + i, args));
 
       // String host = "192.168.1.15" + (i + 1);
       // sites.add(new SeparateBox(host, USER, KEY, new String[] { "init.Boot" }));
     }
 
-    org.junit.runner.JUnitCore.runClasses(KMeansTest.class);
+    // org.junit.runner.JUnitCore.runClasses(KMeansTest.class);
+
+    H2O.main(args);
+
+    Desktop desktop = Desktop.getDesktop();
+    desktop.browse(new URI("http://localhost:54321/Timeline.html"));
+
+    BufferedReader console = new BufferedReader(new InputStreamReader(System.in));
+    console.readLine();
 
-    // H2O.main(new String[] {});
     // TestUtil.stall_till_cloudsize(nodes + 1);
     // new KMeansTest().testGaussian((int) 1e6);