release 0.9.3.1

history.md

@@ -6,6 +6,10 @@
 1. switch apache thrift7 to storm thrift7
 2. set defatult acker number is 1
 3. add "spout.single.thread" setting
+4. make nimbus logview port different from supervisor's
+5. web ui can list all files of log's subdir
+6. Set gc dump dir as log's dir
+
 
 #Release 0.9.3
 ## New feature

jstorm-client-extension/src/main/java/com/alibaba/jstorm/client/ConfigExtension.java

@@ -60,12 +60,28 @@ public static Boolean isNettyEnableDisruptor(Map conf) {
 	 */
 	private static final Integer DEFAULT_DEAMON_HTTPSERVER_PORT = 7621;
 
+	@Deprecated
 	protected static final String DEAMON_HTTPSERVER_PORT = "deamon.logview.port";
 
+	@Deprecated
 	public static Integer getDeamonHttpserverPort(Map conf) {
 		return JStormUtils.parseInt(conf.get(DEAMON_HTTPSERVER_PORT),
 				DEFAULT_DEAMON_HTTPSERVER_PORT);
 	}
+
+	protected static final String SUPERVISOR_DEAMON_HTTPSERVER_PORT = "supervisor.deamon.logview.port";
+
+	public static Integer getSupervisorDeamonHttpserverPort(Map conf) {
+		return JStormUtils.parseInt(conf.get(SUPERVISOR_DEAMON_HTTPSERVER_PORT),
+				DEFAULT_DEAMON_HTTPSERVER_PORT + 1);
+	}
+
+	protected static final String NIMBUS_DEAMON_HTTPSERVER_PORT = "nimbus.deamon.logview.port";
+
+	public static Integer getNimbusDeamonHttpserverPort(Map conf) {
+		return JStormUtils.parseInt(conf.get(NIMBUS_DEAMON_HTTPSERVER_PORT),
+				DEFAULT_DEAMON_HTTPSERVER_PORT);
+	}
 
 	/**
 	 * Worker gc parameter
@@ -418,10 +434,24 @@ public static String getGroupFilePath(Map conf) {
 
 	protected static final String WORKER_GC_PATH = "worker.gc.path";
 
+	/**
+	 * Useless from 0.9.3.1
+	 * 
+	 * @param conf
+	 * @param path
+	 */
+	@Deprecated
 	public static void setWorkerGcPath(Map conf, String path) {
 		conf.put(WORKER_GC_PATH, path);
 	}
 
+	/**
+	 * Useless from 0.9.3.1
+	 * 
+	 * @param conf
+	 * @param path
+	 */
+	@Deprecated
 	public static String getWorkerGcPath(Map conf) {
 		return (String) conf.get(WORKER_GC_PATH);
 	}

jstorm-client-extension/src/main/java/com/alibaba/jstorm/utils/HttpserverUtils.java

@@ -14,6 +14,8 @@ public class HttpserverUtils {
 
 	public static final String HTTPSERVER_LOGVIEW_PARAM_POS = "pos";
 
+	public static final String HTTPSERVER_LOGVIEW_PARAM_DIR = "dir";
+
 	public static final long HTTPSERVER_LOGVIEW_PAGESIZE = 8192;
 
 	public static final String HTTPSERVER_LOGVIEW_PARAM_SIZE_FORMAT = "%016d\n";

jstorm-server/src/main/java/com/alibaba/jstorm/daemon/nimbus/NimbusServer.java

@@ -120,7 +120,8 @@ private void launchServer(final Map conf, INimbus inimbus)  {
 
 			initFollowerThread(conf);
 
-			hs = new Httpserver(conf);
+			int port = ConfigExtension.getNimbusDeamonHttpserverPort(conf);
+			hs = new Httpserver(port);
 			hs.start();
 
 			initContainerHBThread(conf);

jstorm-server/src/main/java/com/alibaba/jstorm/daemon/supervisor/Httpserver.java

@@ -39,11 +39,11 @@ public class Httpserver implements Shutdownable {
 
 	private static Logger LOG = Logger.getLogger(Httpserver.class);
 
-	private static Map storm_conf;
 	private HttpServer hs;
+	private int port;
 
-	public Httpserver(Map storm_conf) {
-		this.storm_conf = storm_conf;
+	public Httpserver(int port) {
+		this.port = port;
 	}
 
 	static class LogHandler implements HttpHandler {
@@ -195,15 +195,23 @@ private Pair<Long, byte[]> queryLog(HttpExchange t,
 
 		}
 
-		byte[] getJSonFiles() throws Exception {
+		byte[] getJSonFiles(String dir) throws Exception {
 			Map<String, FileAttribute> fileMap = new HashMap<String, FileAttribute>();
 
-			File file = new File(logDir);
+
+			String path = logDir;
+			if (dir != null) {
+				path = path + File.separator + dir;
+			}
+
+			LOG.info("List dir " + path);
+
+			File file = new File(path);
 
 			String[] files = file.list();
 
 			for (String fileName : files) {
-				String logFile = Joiner.on(File.separator).join(logDir,
+				String logFile = Joiner.on(File.separator).join(path,
 						fileName);
 
 				FileAttribute fileAttribute = new FileAttribute();
@@ -239,7 +247,8 @@ void handleListDir(HttpExchange t, Map<String, String> paramMap)
 			byte[] filesJson = "Failed to get file list".getBytes();
 
 			try {
-				filesJson = getJSonFiles();
+				String dir = paramMap.get(HttpserverUtils.HTTPSERVER_LOGVIEW_PARAM_DIR);
+				filesJson = getJSonFiles(dir);
 			} catch (Exception e) {
 				handlFailure(t, "Failed to get file list");
 				return;
@@ -255,7 +264,6 @@ void handleListDir(HttpExchange t, Map<String, String> paramMap)
 
 	public void start() {
 		int numHandler = 3;
-		int port = ConfigExtension.getDeamonHttpserverPort(storm_conf);
 		InetSocketAddress socketAddr = new InetSocketAddress(port);
 		Executor executor = Executors.newFixedThreadPool(numHandler);
 

jstorm-server/src/main/java/com/alibaba/jstorm/daemon/supervisor/Supervisor.java

@@ -151,7 +151,8 @@ public SupervisorManger mkSupervisor(Map conf, IContext sharedContext)
 
 
 		//Step 7 start httpserver
-		Httpserver httpserver = new Httpserver(conf);
+		int port = ConfigExtension.getSupervisorDeamonHttpserverPort(conf);
+		Httpserver httpserver = new Httpserver(port);
 		httpserver.start();
 
 

jstorm-server/src/main/java/com/alibaba/jstorm/daemon/supervisor/SyncProcessEvent.java

@@ -482,7 +482,8 @@ public String getChildOpts(Map stormConf) {
 	}
 
 	private String getGcDumpParam(Map totalConf) {
-		String gcPath = ConfigExtension.getWorkerGcPath(totalConf);
+		//String gcPath = ConfigExtension.getWorkerGcPath(totalConf);
+		String gcPath = JStormUtils.getLogDir();
 
 		Date now = new Date();
 		String nowStr = TimeFormat.getSecond(now);

jstorm-server/src/main/resources/defaults.yaml

@@ -96,7 +96,6 @@ supervisor.disk.slot: null
 ### worker.* configs are for task workers
 # worker gc configuration
 # worker.gc.path will put all gc logs and memory dump file
-worker.gc.path: "%JSTORM_HOME%/logs"
 worker.gc.childopts: " -XX:SurvivorRatio=4 -XX:MaxTenuringThreshold=20 -XX:+UseConcMarkSweepGC  -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=70 -XX:CMSFullGCsBeforeCompaction=5 -XX:+HeapDumpOnOutOfMemoryError -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+UseCMSCompactAtFullCollection -XX:CMSMaxAbortablePrecleanTime=5000 "
 worker.heartbeat.frequency.secs: 2
 worker.classpath: ""
@@ -240,4 +239,5 @@ container.heartbeat.frequence: 10
 java.sandbox.enable: false.
 
 # logview port
-deamon.logview.port: 7621
+nimbus.deamon.logview.port: 7621
+supervisor.deamon.logview.port: 7622

jstorm-server/src/test/java/com/alibaba/jstorm/superivosr/deamon/HttpserverTest.java

@@ -2,14 +2,16 @@
 
 import org.junit.Test;
 
+import com.alibaba.jstorm.client.ConfigExtension;
 import com.alibaba.jstorm.daemon.supervisor.Httpserver;
 import com.google.common.collect.Maps;
 
 public class HttpserverTest {
 
 	@Test
 	public void test_httpserver() {
-		Httpserver httpserver = new Httpserver(Maps.newHashMap());
+		int port = ConfigExtension.getSupervisorDeamonHttpserverPort(Maps.newHashMap());
+		Httpserver httpserver = new Httpserver(port);
 		httpserver.start();
 		System.out.println("start....");
 	}

jstorm-ui/src/main/java/com/alibaba/jstorm/ui/UIUtils.java

@@ -38,6 +38,7 @@
 import backtype.storm.utils.NimbusClient;
 import backtype.storm.utils.Utils;
 
+import com.alibaba.jstorm.client.ConfigExtension;
 import com.alibaba.jstorm.common.stats.StatBuckets;
 import com.alibaba.jstorm.common.stats.StaticsType;
 import com.alibaba.jstorm.ui.model.ClusterSumm;
@@ -466,7 +467,7 @@ public static List<TopologySumm> topologySummary(TopologyInfo t) {
 	 * @return
 	 */
 	public static List<ClusterSumm> clusterSummary(ClusterSummary summ,
-			NimbusClient client) throws Exception {
+			NimbusClient client, Map conf) throws Exception {
 		// "Supervisors" "Used slots" "Free slots" "Total slots" "Running task"
 		List<SupervisorSummary> sups = summ.get_supervisors();
 		int supSize = 0;
@@ -533,6 +534,8 @@ public static List<ClusterSumm> clusterSummary(ClusterSummary summ,
 		} else {
 			clusterSumm.setNimbusIp(NetWorkUtils.host2Ip(master));
 		}
+		int port = ConfigExtension.getNimbusDeamonHttpserverPort(conf);
+		clusterSumm.setNimbusLogPort(String.valueOf(port));
 		if (summ.is_isGroupModel())
 			clusterSumm.setIsGroupModel("true");
 		else

jstorm-ui/src/main/java/com/alibaba/jstorm/ui/model/ClusterSumm.java

@@ -11,6 +11,7 @@ public class ClusterSumm implements Serializable {
 	private static final long serialVersionUID = -7936384220562644886L;
 
 	private String nimbusIp;
+	private String nimbusLogPort;
 	private String nimbusHostname;
 	private String nimbusUptime;
 	private String isGroupModel;
@@ -37,6 +38,14 @@ public void setNimbusIp(String nimbusIp) {
 		this.nimbusIp = nimbusIp;
 	}
 
+	public String getNimbusLogPort() {
+		return nimbusLogPort;
+	}
+
+	public void setNimbusLogPort(String nimbusLogPort) {
+		this.nimbusLogPort = nimbusLogPort;
+	}
+
 	public String getNimbusHostname() {
 		return nimbusHostname;
 	}

jstorm-ui/src/main/java/com/alibaba/jstorm/ui/model/NimbusSlave.java

@@ -12,15 +12,18 @@ public class NimbusSlave implements Serializable {
 
 	private String ip;
 
+	private String logviewPort;
+
 	private String hostname;
 
 	private String uptime;
 
-	public NimbusSlave(String hostname, String uptime) {
+	public NimbusSlave(String hostname, String uptime, int port) {
 		this.hostname = hostname;
 		this.uptime = uptime;
 		String[] fields = StringUtils.split(hostname, ":");
 		this.ip = NetWorkUtils.host2Ip(fields[0]);
+		this.logviewPort = String.valueOf(port);
 	}
 
 	public String getIp() {
@@ -43,5 +46,14 @@ public String getUptime() {
 	public void setUptime(String uptime) {
 		this.uptime = uptime;
 	}
+
+	public String getLogviewPort() {
+		return logviewPort;
+	}
+
+	public void setLogviewPort(String logviewPort) {
+		this.logviewPort = logviewPort;
+	}
+
 
 }