| /** |
| * 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 com.alibaba.jstorm.utils; |
| |
| import java.io.BufferedOutputStream; |
| import java.io.ByteArrayOutputStream; |
| import java.io.File; |
| import java.io.FileOutputStream; |
| import java.io.IOException; |
| import java.io.PrintStream; |
| import java.io.PrintWriter; |
| import java.io.StringWriter; |
| import java.lang.management.ManagementFactory; |
| import java.lang.management.MemoryMXBean; |
| import java.lang.management.MemoryUsage; |
| import java.math.BigInteger; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Enumeration; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Set; |
| import java.util.TreeMap; |
| import java.util.concurrent.TimeUnit; |
| import java.util.zip.ZipEntry; |
| import java.util.zip.ZipFile; |
| |
| import javax.management.ObjectName; |
| |
| import org.apache.commons.exec.CommandLine; |
| import org.apache.commons.exec.DefaultExecutor; |
| import org.apache.commons.exec.ExecuteException; |
| import org.apache.commons.exec.ExecuteResultHandler; |
| import org.apache.commons.exec.PumpStreamHandler; |
| import org.apache.commons.lang.StringUtils; |
| import org.apache.thrift.TFieldIdEnum; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import backtype.storm.Config; |
| import backtype.storm.utils.Utils; |
| |
| import com.alibaba.jstorm.callback.AsyncLoopDefaultKill; |
| import com.alibaba.jstorm.callback.RunnableCallback; |
| import com.alibaba.jstorm.client.ConfigExtension; |
| |
| /** |
| * JStorm utility |
| * |
| * @author yannian/Longda/Xin.Zhou/Xin.Li |
| */ |
| public class JStormUtils { |
| private static final Logger LOG = LoggerFactory.getLogger(JStormUtils.class); |
| |
| public static long SIZE_1_K = 1024; |
| public static long SIZE_1_M = SIZE_1_K * 1024; |
| public static long SIZE_1_G = SIZE_1_M * 1024; |
| public static long SIZE_1_T = SIZE_1_G * 1024; |
| public static long SIZE_1_P = SIZE_1_T * 1024; |
| |
| public static final int MIN_1 = 60; |
| public static final int MIN_30 = MIN_1 * 30; |
| public static final int HOUR_1 = MIN_30 * 2; |
| public static final int DAY_1 = HOUR_1 * 24; |
| |
| public static String getErrorInfo(String baseInfo, Exception e) { |
| try { |
| StringWriter sw = new StringWriter(); |
| PrintWriter pw = new PrintWriter(sw); |
| e.printStackTrace(pw); |
| return baseInfo + "\r\n" + sw.toString() + "\r\n"; |
| } catch (Exception e2) { |
| return baseInfo; |
| } |
| } |
| |
| public static String getErrorInfo(Throwable error) { |
| try { |
| StringWriter sw = new StringWriter(); |
| PrintWriter pw = new PrintWriter(sw); |
| error.printStackTrace(pw); |
| return sw.toString(); |
| } catch (Exception e1) { |
| return ""; |
| } |
| } |
| |
| /** |
| * filter the map |
| * |
| * @param filter |
| * @param all |
| * @return |
| */ |
| public static <K, V> Map<K, V> select_keys_pred(Set<K> filter, Map<K, V> all) { |
| Map<K, V> filterMap = new HashMap<K, V>(); |
| |
| for (Entry<K, V> entry : all.entrySet()) { |
| if (!filter.contains(entry.getKey())) { |
| filterMap.put(entry.getKey(), entry.getValue()); |
| } |
| } |
| |
| return filterMap; |
| } |
| |
| public static byte[] barr(byte v) { |
| byte[] byteArray = new byte[1]; |
| byteArray[0] = v; |
| |
| return byteArray; |
| } |
| |
| public static byte[] barr(Short v) { |
| byte[] byteArray = new byte[Short.SIZE / 8]; |
| for (int i = 0; i < byteArray.length; i++) { |
| int off = (byteArray.length - 1 - i) * 8; |
| byteArray[i] = (byte) ((v >> off) & 0xFF); |
| } |
| return byteArray; |
| } |
| |
| public static byte[] barr(Integer v) { |
| byte[] byteArray = new byte[Integer.SIZE / 8]; |
| for (int i = 0; i < byteArray.length; i++) { |
| int off = (byteArray.length - 1 - i) * 8; |
| byteArray[i] = (byte) ((v >> off) & 0xFF); |
| } |
| return byteArray; |
| } |
| |
| // for test |
| public static int byteToInt2(byte[] b) { |
| |
| int iOutcome = 0; |
| byte bLoop; |
| |
| for (int i = 0; i < 4; i++) { |
| bLoop = b[i]; |
| int off = (b.length - 1 - i) * 8; |
| iOutcome += (bLoop & 0xFF) << off; |
| |
| } |
| |
| return iOutcome; |
| } |
| |
| /** |
| * LocalMode variable isn't clean, it make the JStormUtils ugly |
| */ |
| public static boolean localMode = false; |
| |
| public static boolean isLocalMode() { |
| return localMode; |
| } |
| |
| public static void setLocalMode(boolean localMode) { |
| JStormUtils.localMode = localMode; |
| } |
| |
| public static void haltProcess(int val) { |
| Runtime.getRuntime().halt(val); |
| } |
| |
| public static void halt_process(int val, String msg) { |
| LOG.info("Halting process: " + msg); |
| try { |
| Thread.sleep(1000); |
| } catch (InterruptedException e) { |
| } |
| if (localMode && val == 0) { |
| // throw new RuntimeException(msg); |
| } else { |
| haltProcess(val); |
| } |
| } |
| |
| /** |
| * "{:a 1 :b 1 :c 2} -> {1 [:a :b] 2 :c}" |
| * |
| * @param map |
| * @return |
| */ |
| public static <K, V> HashMap<V, List<K>> reverse_map(Map<K, V> map) { |
| HashMap<V, List<K>> rtn = new HashMap<V, List<K>>(); |
| if (map == null) { |
| return rtn; |
| } |
| for (Entry<K, V> entry : map.entrySet()) { |
| K key = entry.getKey(); |
| V val = entry.getValue(); |
| List<K> list = rtn.get(val); |
| if (list == null) { |
| list = new ArrayList<K>(); |
| rtn.put(entry.getValue(), list); |
| } |
| list.add(key); |
| |
| } |
| |
| return rtn; |
| } |
| |
| /** |
| * Gets the pid of this JVM, because Java doesn't provide a real way to do this. |
| * |
| * @return |
| */ |
| public static String process_pid() { |
| String name = ManagementFactory.getRuntimeMXBean().getName(); |
| String[] split = name.split("@"); |
| if (split.length != 2) { |
| throw new RuntimeException("Got unexpected process name: " + name); |
| } |
| |
| return split[0]; |
| } |
| |
| public static void exec_command(String command) throws ExecuteException, IOException { |
| String[] cmdlist = command.split(" "); |
| CommandLine cmd = new CommandLine(cmdlist[0]); |
| for (int i = 1; i < cmdlist.length; i++) { |
| cmd.addArgument(cmdlist[i]); |
| } |
| |
| DefaultExecutor exec = new DefaultExecutor(); |
| exec.execute(cmd); |
| } |
| |
| /** |
| * Extra dir from the jar to destdir |
| * |
| * @param jarpath |
| * @param dir |
| * @param destdir |
| */ |
| public static void extract_dir_from_jar(String jarpath, String dir, String destdir) { |
| String cmd = "unzip -qq " + jarpath + " " + dir + "/** -d " + destdir; |
| try { |
| exec_command(cmd); |
| } catch (Exception e) { |
| LOG.warn("No " + dir + " from " + jarpath + " by cmd:" + cmd + "!\n" + e.getMessage()); |
| } |
| |
| } |
| |
| public static void ensure_process_killed(Integer pid) { |
| // in this function, just kill the process 5 times |
| // make sure the process be killed definitely |
| for (int i = 0; i < 5; i++) { |
| try { |
| exec_command("kill -9 " + pid); |
| LOG.info("kill -9 process " + pid); |
| sleepMs(100); |
| } catch (ExecuteException e) { |
| LOG.info("Error when trying to kill " + pid + ". Process has been killed"); |
| } catch (Exception e) { |
| LOG.info("Error when trying to kill " + pid + ".Exception ", e); |
| } |
| } |
| } |
| |
| public static void process_killed(Integer pid) { |
| try { |
| exec_command("kill " + pid); |
| LOG.info("kill process " + pid); |
| } catch (ExecuteException e) { |
| LOG.info("Error when trying to kill " + pid + ". Process has been killed. "); |
| } catch (Exception e) { |
| LOG.info("Error when trying to kill " + pid + ".Exception ", e); |
| } |
| } |
| |
| public static void kill(Integer pid) { |
| process_killed(pid); |
| |
| sleepMs(5 * 1000); |
| |
| ensure_process_killed(pid); |
| } |
| |
| public static void kill_signal(Integer pid, String signal) { |
| String cmd = "kill " + signal + " " + pid; |
| try { |
| exec_command(cmd); |
| LOG.info(cmd); |
| } catch (ExecuteException e) { |
| LOG.info("Error when run " + cmd + ". Process has been killed. "); |
| } catch (Exception e) { |
| LOG.info("Error when run " + cmd + ". Exception ", e); |
| } |
| } |
| |
| /** |
| * This function is only for linux |
| * |
| * @param pid |
| * @return |
| */ |
| public static boolean isProcDead(String pid) { |
| if (OSInfo.isLinux() == false) { |
| return false; |
| } |
| |
| String path = "/proc/" + pid; |
| File file = new File(path); |
| |
| if (file.exists() == false) { |
| LOG.info("Process " + pid + " is dead"); |
| return true; |
| } |
| |
| return false; |
| } |
| |
| public static Double getCpuUsage() { |
| if (OSInfo.isLinux() == false) { |
| return 0.0; |
| } |
| |
| Double value = 0.0; |
| String output = null; |
| try { |
| String pid = JStormUtils.process_pid(); |
| String command = String.format("top -b -n 1 -p %s | grep -w %s", pid, pid); |
| output = SystemOperation.exec(command); |
| String subStr = output.substring(output.indexOf("S") + 1); |
| for (int i = 0; i < subStr.length(); i++) { |
| char ch = subStr.charAt(i); |
| if (ch != ' ') { |
| subStr = subStr.substring(i); |
| break; |
| } |
| } |
| String usedCpu = subStr.substring(0, subStr.indexOf(" ")); |
| value = Double.valueOf(usedCpu); |
| } catch (Exception e) { |
| LOG.warn("Failed to get cpu usage ratio."); |
| if (output != null) |
| LOG.warn("Output string is \"" + output + "\""); |
| value = 0.0; |
| } |
| |
| return value; |
| } |
| |
| public static Double getDiskUsage() { |
| if (!OSInfo.isLinux() && !OSInfo.isMac()) { |
| return 0.0; |
| } |
| try { |
| String output = SystemOperation.exec("df -h /"); |
| if (output != null) { |
| String[] lines = output.split("[\\r\\n]+"); |
| if (lines.length >= 2) { |
| String[] parts = lines[1].split("\\s+"); |
| if (parts.length >= 5) { |
| String pct = parts[4]; |
| if (pct.endsWith("%")) { |
| return Integer.valueOf(pct.substring(0, pct.length() - 1)) / 100.0; |
| } |
| } |
| } |
| } |
| } catch (Exception e) { |
| LOG.warn("failed to get disk usage:", e); |
| } |
| return 0.0; |
| } |
| |
| public static Double getMemUsage() { |
| if (OSInfo.isLinux() == true) { |
| try { |
| Double value = 0.0; |
| String pid = JStormUtils.process_pid(); |
| String command = String.format("top -b -n 1 -p %s | grep -w %s", pid, pid); |
| String output = SystemOperation.exec(command); |
| |
| int m = 0; |
| String[] strArray = output.split(" "); |
| for (int i = 0; i < strArray.length; i++) { |
| String info = strArray[i]; |
| if (info.trim().length() == 0) { |
| continue; |
| } |
| if (m == 5) { |
| // memory |
| String unit = info.substring(info.length() - 1); |
| |
| if (unit.equalsIgnoreCase("g")) { |
| value = Double.parseDouble(info.substring(0, info.length() - 1)); |
| value *= 1000000000; |
| } else if (unit.equalsIgnoreCase("m")) { |
| value = Double.parseDouble(info.substring(0, info.length() - 1)); |
| value *= 1000000; |
| } else { |
| value = Double.parseDouble(info); |
| } |
| |
| //LOG.info("!!!! Get Memory Size:{}, info:{}", value, info); |
| return value; |
| } |
| if (m == 8) { |
| // cpu usage |
| |
| } |
| if (m == 9) { |
| // memory ratio |
| |
| } |
| m++; |
| } |
| } catch (Exception e) { |
| LOG.warn("Failed to get memory usage ."); |
| |
| } |
| } |
| |
| // this will be incorrect |
| MemoryMXBean memoryMXBean = ManagementFactory.getMemoryMXBean(); |
| MemoryUsage memoryUsage = memoryMXBean.getHeapMemoryUsage(); |
| |
| return Double.valueOf(memoryUsage.getUsed()); |
| } |
| |
| /** |
| * If it is backend, please set resultHandler, such as DefaultExecuteResultHandler If it is frontend, ByteArrayOutputStream.toString get the result |
| * <p/> |
| * This function don't care whether the command is successfully or not |
| * |
| * @param command |
| * @param environment |
| * @param workDir |
| * @param resultHandler |
| * @return |
| * @throws IOException |
| */ |
| public static ByteArrayOutputStream launchProcess(String command, final Map environment, final String workDir, ExecuteResultHandler resultHandler) |
| throws IOException { |
| |
| String[] cmdlist = command.split(" "); |
| |
| CommandLine cmd = new CommandLine(cmdlist[0]); |
| for (String cmdItem : cmdlist) { |
| if (StringUtils.isBlank(cmdItem) == false) { |
| cmd.addArgument(cmdItem); |
| } |
| } |
| |
| DefaultExecutor executor = new DefaultExecutor(); |
| |
| executor.setExitValue(0); |
| if (StringUtils.isBlank(workDir) == false) { |
| executor.setWorkingDirectory(new File(workDir)); |
| } |
| |
| ByteArrayOutputStream out = new ByteArrayOutputStream(); |
| |
| PumpStreamHandler streamHandler = new PumpStreamHandler(out, out); |
| if (streamHandler != null) { |
| executor.setStreamHandler(streamHandler); |
| } |
| |
| try { |
| if (resultHandler == null) { |
| executor.execute(cmd, environment); |
| } else { |
| executor.execute(cmd, environment, resultHandler); |
| } |
| } catch (ExecuteException e) { |
| |
| // @@@@ |
| // failed to run command |
| } |
| |
| return out; |
| |
| } |
| |
| protected static Process launchProcess(final String[] cmdlist, final Map<String, String> environment) throws IOException { |
| ArrayList<String> buff = new ArrayList<String>(); |
| for (String tok : cmdlist) { |
| if (!tok.isEmpty()) { |
| buff.add(tok); |
| } |
| } |
| |
| ProcessBuilder builder = new ProcessBuilder(buff); |
| builder.redirectErrorStream(true); |
| Map<String, String> process_evn = builder.environment(); |
| for (Entry<String, String> entry : environment.entrySet()) { |
| process_evn.put(entry.getKey(), entry.getValue()); |
| } |
| |
| return builder.start(); |
| } |
| |
| /** |
| * @param command |
| * @param environment |
| * @param backend |
| * @return |
| * @throws IOException |
| * @@@ it should use DefaultExecutor to start a process, but some little problem have been found, such as exitCode/output string so still use the old method |
| * to start process |
| */ |
| public static Process launch_process(final String command, final Map<String, String> environment, boolean backend) throws IOException { |
| |
| if (backend == true) { |
| new Thread(new Runnable() { |
| |
| @Override |
| public void run() { |
| String[] cmdlist = (new String("nohup " + command + " &")).split(" "); |
| try { |
| launchProcess(cmdlist, environment); |
| } catch (IOException e) { |
| LOG.error("Failed to run " + command + ":" + e.getCause(), e); |
| } |
| } |
| }).start(); |
| return null; |
| } else { |
| String[] cmdlist = command.split(" "); |
| return launchProcess(cmdlist, environment); |
| } |
| } |
| |
| public static String current_classpath() { |
| return System.getProperty("java.class.path"); |
| } |
| |
| // public static String add_to_classpath(String classpath, String[] paths) { |
| // for (String path : paths) { |
| // classpath += ":" + path; |
| // } |
| // return classpath; |
| // } |
| |
| public static String to_json(Map m) { |
| return Utils.to_json(m); |
| } |
| |
| public static Object from_json(String json) { |
| return Utils.from_json(json); |
| } |
| |
| public static <V> HashMap<V, Integer> multi_set(List<V> list) { |
| HashMap<V, Integer> rtn = new HashMap<V, Integer>(); |
| for (V v : list) { |
| int cnt = 1; |
| if (rtn.containsKey(v)) { |
| cnt += rtn.get(v); |
| } |
| rtn.put(v, cnt); |
| } |
| return rtn; |
| } |
| |
| /** |
| * if the list exist repeat string, return the repeated string |
| * <p/> |
| * this function will be used to check wheter bolt or spout exist same id |
| * |
| * @param sets |
| * @return |
| */ |
| public static List<String> getRepeat(List<String> list) { |
| |
| List<String> rtn = new ArrayList<String>(); |
| Set<String> idSet = new HashSet<String>(); |
| |
| for (String id : list) { |
| if (idSet.contains(id)) { |
| rtn.add(id); |
| } else { |
| idSet.add(id); |
| } |
| } |
| |
| return rtn; |
| } |
| |
| /** |
| * balance all T |
| * |
| * @param <T> |
| * @param splitup |
| * @return |
| */ |
| public static <T> List<T> interleave_all(List<List<T>> splitup) { |
| ArrayList<T> rtn = new ArrayList<T>(); |
| int maxLength = 0; |
| for (List<T> e : splitup) { |
| int len = e.size(); |
| if (maxLength < len) { |
| maxLength = len; |
| } |
| } |
| |
| for (int i = 0; i < maxLength; i++) { |
| for (List<T> e : splitup) { |
| if (e.size() > i) { |
| rtn.add(e.get(i)); |
| } |
| } |
| } |
| |
| return rtn; |
| } |
| |
| public static Long bit_xor_vals(Object... vals) { |
| Long rtn = 0l; |
| for (Object n : vals) { |
| rtn = bit_xor(rtn, n); |
| } |
| |
| return rtn; |
| } |
| |
| public static <T> Long bit_xor_vals(List<T> vals) { |
| Long rtn = 0l; |
| for (T n : vals) { |
| rtn = bit_xor(rtn, n); |
| } |
| |
| return rtn; |
| } |
| |
| public static <T> Long bit_xor_vals_sets(Set<T> vals) { |
| Long rtn = 0l; |
| for (T n : vals) { |
| rtn = bit_xor(rtn, n); |
| } |
| return rtn; |
| } |
| |
| public static Long bit_xor(Object a, Object b) { |
| Long rtn = 0l; |
| |
| if (a instanceof Long && b instanceof Long) { |
| rtn = ((Long) a) ^ ((Long) b); |
| return rtn; |
| } else if (b instanceof Set) { |
| Long bs = bit_xor_vals_sets((Set) b); |
| return bit_xor(a, bs); |
| } else if (a instanceof Set) { |
| Long as = bit_xor_vals_sets((Set) a); |
| return bit_xor(as, b); |
| } else { |
| Long ai = Long.parseLong(String.valueOf(a)); |
| Long bi = Long.parseLong(String.valueOf(b)); |
| rtn = ai ^ bi; |
| return rtn; |
| } |
| |
| } |
| |
| public static <V> List<V> mk_list(V... args) { |
| ArrayList<V> rtn = new ArrayList<V>(); |
| for (V o : args) { |
| rtn.add(o); |
| } |
| return rtn; |
| } |
| |
| public static <V> List<V> mk_list(Set<V> args) { |
| ArrayList<V> rtn = new ArrayList<V>(); |
| if (args != null) { |
| for (V o : args) { |
| rtn.add(o); |
| } |
| } |
| return rtn; |
| } |
| |
| public static <V> List<V> mk_list(Collection<V> args) { |
| ArrayList<V> rtn = new ArrayList<V>(); |
| if (args != null) { |
| for (V o : args) { |
| rtn.add(o); |
| } |
| } |
| return rtn; |
| } |
| |
| public static <V> V[] mk_arr(V... args) { |
| return args; |
| } |
| |
| public static Long parseLong(Object o) { |
| if (o == null) { |
| return null; |
| } |
| |
| if (o instanceof String) { |
| return Long.valueOf(String.valueOf(o)); |
| } else if (o instanceof Integer) { |
| Integer value = (Integer) o; |
| return Long.valueOf((Integer) value); |
| } else if (o instanceof Long) { |
| return (Long) o; |
| } else { |
| throw new RuntimeException("Invalid value " + o.getClass().getName() + " " + o); |
| } |
| } |
| |
| public static Double parseDouble(Object o) { |
| if (o == null) { |
| return null; |
| } |
| |
| if (o instanceof String) { |
| return Double.valueOf(String.valueOf(o)); |
| } else if (o instanceof Integer) { |
| Number value = (Integer) o; |
| return value.doubleValue(); |
| } else if (o instanceof Long) { |
| Number value = (Long) o; |
| return value.doubleValue(); |
| } else if (o instanceof Double) { |
| return (Double) o; |
| } else { |
| throw new RuntimeException("Invalid value " + o.getClass().getName() + " " + o); |
| } |
| } |
| |
| public static Double parseDouble(Object o, double defaultValue) { |
| if (o == null) { |
| return defaultValue; |
| } |
| try { |
| return parseDouble(o); |
| } catch (Exception ignored) { |
| return defaultValue; |
| } |
| } |
| |
| public static Long parseLong(Object o, long defaultValue) { |
| |
| if (o == null) { |
| return defaultValue; |
| } |
| |
| if (o instanceof String) { |
| return Long.valueOf(String.valueOf(o)); |
| } else if (o instanceof Integer) { |
| Integer value = (Integer) o; |
| return Long.valueOf((Integer) value); |
| } else if (o instanceof Long) { |
| return (Long) o; |
| } else { |
| return defaultValue; |
| } |
| } |
| |
| public static Integer parseInt(Object o) { |
| if (o == null) { |
| return null; |
| } |
| |
| if (o instanceof String) { |
| return Integer.parseInt(String.valueOf(o)); |
| } else if (o instanceof Long) { |
| long value = (Long) o; |
| return Integer.valueOf((int) value); |
| } else if (o instanceof Integer) { |
| return (Integer) o; |
| } else { |
| throw new RuntimeException("Invalid value " + o.getClass().getName() + " " + o); |
| } |
| } |
| |
| public static Integer parseInt(Object o, int defaultValue) { |
| if (o == null) { |
| return defaultValue; |
| } |
| |
| if (o instanceof String) { |
| return Integer.parseInt(String.valueOf(o)); |
| } else if (o instanceof Long) { |
| long value = (Long) o; |
| return Integer.valueOf((int) value); |
| } else if (o instanceof Integer) { |
| return (Integer) o; |
| } else { |
| return defaultValue; |
| } |
| } |
| |
| public static Boolean parseBoolean(Object o) { |
| if (o == null) { |
| return null; |
| } |
| |
| if (o instanceof String) { |
| return Boolean.valueOf((String) o); |
| } else if (o instanceof Boolean) { |
| return (Boolean) o; |
| } else { |
| throw new RuntimeException("Invalid value " + o.getClass().getName() + " " + o); |
| } |
| } |
| |
| public static boolean parseBoolean(Object o, boolean defaultValue) { |
| if (o == null) { |
| return defaultValue; |
| } |
| |
| if (o instanceof String) { |
| return Boolean.valueOf((String) o); |
| } else if (o instanceof Boolean) { |
| return (Boolean) o; |
| } else { |
| return defaultValue; |
| } |
| } |
| |
| public static <V> Set<V> listToSet(List<V> list) { |
| if (list == null) { |
| return null; |
| } |
| |
| Set<V> set = new HashSet<V>(); |
| set.addAll(list); |
| return set; |
| } |
| |
| /** |
| * Check whether the zipfile contain the resources |
| * |
| * @param zipfile |
| * @param resources |
| * @return |
| */ |
| public static boolean zipContainsDir(String zipfile, String resources) { |
| |
| Enumeration<? extends ZipEntry> entries = null; |
| try { |
| entries = (new ZipFile(zipfile)).entries(); |
| while (entries != null && entries.hasMoreElements()) { |
| ZipEntry ze = entries.nextElement(); |
| String name = ze.getName(); |
| if (name.startsWith(resources + "/")) { |
| return true; |
| } |
| } |
| } catch (IOException e) { |
| // TODO Auto-generated catch block |
| // e.printStackTrace(); |
| LOG.error(e + "zipContainsDir error"); |
| } |
| |
| return false; |
| } |
| |
| public static Object add(Object oldValue, Object newValue) { |
| if (oldValue == null) { |
| return newValue; |
| } else if (newValue == null) { |
| return oldValue; |
| } |
| |
| if (oldValue instanceof Long) { |
| return (Long) oldValue + (Long) newValue; |
| } else if (oldValue instanceof Double) { |
| return (Double) oldValue + (Double) newValue; |
| } else if (oldValue instanceof Integer) { |
| return (Integer) oldValue + (Integer) newValue; |
| } else if (oldValue instanceof Float) { |
| return (Float) oldValue + (Float) newValue; |
| } else if (oldValue instanceof Short) { |
| return (Short) oldValue + (Short) newValue; |
| } else if (oldValue instanceof BigInteger) { |
| return ((BigInteger) oldValue).add((BigInteger) newValue); |
| } else if (oldValue instanceof Number) { |
| return ((Number) oldValue).doubleValue() + ((Number) newValue).doubleValue(); |
| } else { |
| return null; |
| } |
| } |
| |
| public static Object mergeList(List<Object> list) { |
| Object ret = null; |
| |
| for (Object value : list) { |
| ret = add(ret, value); |
| } |
| |
| return ret; |
| } |
| |
| public static List<Object> mergeList(List<Object> result, Object add) { |
| if (add instanceof Collection) { |
| for (Object o : (Collection) add) { |
| result.add(o); |
| } |
| } else if (add instanceof Set) { |
| for (Object o : (Collection) add) { |
| result.add(o); |
| } |
| } else { |
| result.add(add); |
| } |
| |
| return result; |
| } |
| |
| public static List<Object> distinctList(List<Object> input) { |
| List<Object> retList = new ArrayList<Object>(); |
| |
| for (Object object : input) { |
| if (retList.contains(object)) { |
| continue; |
| } else { |
| retList.add(object); |
| } |
| |
| } |
| |
| return retList; |
| } |
| |
| public static <K, V> Map<K, V> mergeMapList(List<Map<K, V>> list) { |
| Map<K, V> ret = new HashMap<K, V>(); |
| |
| for (Map<K, V> listEntry : list) { |
| if (listEntry == null) { |
| continue; |
| } |
| for (Entry<K, V> mapEntry : listEntry.entrySet()) { |
| K key = mapEntry.getKey(); |
| V value = mapEntry.getValue(); |
| |
| V retValue = (V) add(ret.get(key), value); |
| |
| ret.put(key, retValue); |
| } |
| } |
| |
| return ret; |
| } |
| |
| public static String formatSimpleDouble(Double value) { |
| try { |
| java.text.DecimalFormat form = new java.text.DecimalFormat("##0.000"); |
| String s = form.format(value); |
| return s; |
| } catch (Exception e) { |
| return "0.000"; |
| } |
| |
| } |
| |
| public static double formatDoubleDecPoint2(Double value) { |
| try { |
| java.text.DecimalFormat form = new java.text.DecimalFormat("##.00"); |
| String s = form.format(value); |
| return Double.valueOf(s); |
| } catch (Exception e) { |
| return 0.0; |
| } |
| } |
| |
| public static double formatDoubleDecPoint4(Double value) { |
| try { |
| java.text.DecimalFormat form = new java.text.DecimalFormat("###.0000"); |
| String s = form.format(value); |
| return Double.valueOf(s); |
| } catch (Exception e) { |
| return 0.0; |
| } |
| } |
| |
| public static Double convertToDouble(Object value) { |
| Double ret; |
| |
| if (value == null) { |
| ret = null; |
| } else { |
| if (value instanceof Integer) { |
| ret = ((Integer) value).doubleValue(); |
| } else if (value instanceof Long) { |
| ret = ((Long) value).doubleValue(); |
| } else if (value instanceof Float) { |
| ret = ((Float) value).doubleValue(); |
| } else if (value instanceof Double) { |
| ret = (Double) value; |
| } else { |
| ret = null; |
| } |
| } |
| |
| return ret; |
| } |
| |
| public static String formatValue(Object value) { |
| if (value == null) { |
| return "0"; |
| } |
| |
| if (value instanceof Long) { |
| return String.valueOf((Long) value); |
| } else if (value instanceof Double) { |
| return formatSimpleDouble((Double) value); |
| } else { |
| return String.valueOf(value); |
| } |
| } |
| |
| public static void sleepMs(long ms) { |
| try { |
| Thread.sleep(ms); |
| } catch (InterruptedException e) { |
| |
| } |
| } |
| |
| public static void sleepNs(int ns) { |
| try { |
| Thread.sleep(0, ns); |
| } catch (InterruptedException e) { |
| |
| } |
| } |
| |
| public static String HEXES = "0123456789ABCDEF"; |
| |
| public static String toPrintableString(byte[] buf) { |
| if (buf == null) { |
| return null; |
| } |
| |
| StringBuilder sb = new StringBuilder(); |
| int index = 0; |
| for (byte b : buf) { |
| if (index % 10 == 0) { |
| sb.append("\n"); |
| } |
| index++; |
| |
| sb.append(HEXES.charAt((b & 0xF0) >> 4)); |
| sb.append(HEXES.charAt((b & 0x0F))); |
| sb.append(" "); |
| |
| } |
| |
| return sb.toString(); |
| } |
| |
| /** |
| * @return |
| * @@@ Todo |
| */ |
| public static Long getPhysicMemorySize() { |
| Object object; |
| try { |
| object = ManagementFactory.getPlatformMBeanServer().getAttribute(new ObjectName("java.lang", "type", "OperatingSystem"), "TotalPhysicalMemorySize"); |
| } catch (Exception e) { |
| LOG.warn("Failed to get system physical memory size,", e); |
| return null; |
| } |
| |
| Long ret = (Long) object; |
| |
| return ret; |
| } |
| |
| public static String genLogName(String topology, Integer port) { |
| return topology + "-worker-" + port + ".log"; |
| } |
| |
| // public static String getLog4jFileName(org.apache.log4j.Logger |
| // log4jLogger) throws Exception{ |
| // Enumeration<org.apache.log4j.Appender> enumAppender = |
| // log4jLogger.getAllAppenders(); |
| // org.apache.log4j.FileAppender fileAppender = null; |
| // while (enumAppender.hasMoreElements()) { |
| // org.apache.log4j.Appender appender = enumAppender.nextElement(); |
| // if (appender instanceof org.apache.log4j.FileAppender) { |
| // fileAppender = (org.apache.log4j.FileAppender) appender; |
| // break; |
| // } |
| // } |
| // if (fileAppender != null) { |
| // return fileAppender.getFile(); |
| // |
| // } |
| // |
| // return null; |
| // } |
| |
| public static String getLogFileName() { |
| try { |
| Logger rootLogger = LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME); |
| if (rootLogger instanceof ch.qos.logback.classic.Logger) { |
| ch.qos.logback.classic.Logger logbackLogger = (ch.qos.logback.classic.Logger) rootLogger; |
| // Logger framework is Logback |
| for (Iterator<ch.qos.logback.core.Appender<ch.qos.logback.classic.spi.ILoggingEvent>> index = logbackLogger.iteratorForAppenders(); index |
| .hasNext();) { |
| ch.qos.logback.core.Appender<ch.qos.logback.classic.spi.ILoggingEvent> appender = index.next(); |
| if (appender instanceof ch.qos.logback.core.FileAppender) { |
| ch.qos.logback.core.FileAppender fileAppender = (ch.qos.logback.core.FileAppender) appender; |
| return fileAppender.getFile(); |
| } |
| } |
| } |
| // else if (rootLogger instanceof org.slf4j.impl.Log4jLoggerAdapter) |
| // { |
| // // slf4j-log4j |
| // org.slf4j.impl.Log4jLoggerAdapter log4jAdapter = |
| // (org.slf4j.impl.Log4jLoggerAdapter) rootLogger; |
| // try { |
| // Field field = log4jAdapter.getClass().getDeclaredField( |
| // "logger"); |
| // field.setAccessible(true); |
| // Object obj = field.get(log4jAdapter); |
| // if (obj instanceof org.apache.log4j.spi.RootLogger) { |
| // return getLog4jFileName((org.apache.log4j.spi.RootLogger) obj); |
| // } |
| // } catch (Exception e) { |
| // e.printStackTrace(); |
| // } |
| // |
| // } else if (rootLogger instanceof org.apache.log4j.Logger) { |
| // return getLog4jFileName((org.apache.log4j.Logger) rootLogger); |
| // } |
| |
| } catch (Throwable e) { |
| LOG.info("Failed to get root logger file name", e.getMessage()); |
| return null; |
| } |
| return null; |
| } |
| |
| public static String getLogDir() { |
| String file = JStormUtils.getLogFileName(); |
| if (file != null) { |
| if (file.lastIndexOf(File.separator) < 0) |
| return ""; |
| return file.substring(0, file.lastIndexOf(File.separator)); |
| } |
| |
| String stormHome = System.getProperty("jstorm.home"); |
| if (stormHome == null) { |
| return "." + File.separator + "logs"; |
| } else { |
| return stormHome + File.separator + "logs"; |
| } |
| } |
| |
| public static void redirectOutput(String file) throws Exception { |
| |
| System.out.println("Redirect output to " + file); |
| |
| FileOutputStream workerOut = new FileOutputStream(new File(file)); |
| |
| PrintStream ps = new PrintStream(new BufferedOutputStream(workerOut), true); |
| System.setOut(ps); |
| System.setErr(ps); |
| |
| LOG.info("Successfully redirect System.out to " + file); |
| |
| } |
| |
| public static RunnableCallback getDefaultKillfn() { |
| |
| return new AsyncLoopDefaultKill(); |
| } |
| |
| public static TreeMap<Integer, Integer> integer_divided(int sum, int num_pieces) { |
| return Utils.integerDivided(sum, num_pieces); |
| } |
| |
| public static <K, V> HashMap<K, V> filter_val(RunnableCallback fn, Map<K, V> amap) { |
| HashMap<K, V> rtn = new HashMap<K, V>(); |
| |
| for (Entry<K, V> entry : amap.entrySet()) { |
| V value = entry.getValue(); |
| Object result = fn.execute(value); |
| |
| if (result == (Boolean) true) { |
| rtn.put(entry.getKey(), value); |
| } |
| } |
| return rtn; |
| } |
| |
| public static List<Integer> getSupervisorPortList(Map conf) { |
| List<Integer> portList = (List<Integer>) conf.get(Config.SUPERVISOR_SLOTS_PORTS); |
| if (portList != null && portList.size() > 0) { |
| return portList; |
| } |
| |
| LOG.info("Generate port list through CPU cores and system memory size"); |
| |
| double cpuWeight = ConfigExtension.getSupervisorSlotsPortCpuWeight(conf); |
| int sysCpuNum = 4; |
| try { |
| sysCpuNum = Runtime.getRuntime().availableProcessors(); |
| } catch (Exception e) { |
| LOG.info("Failed to get CPU cores, set cpu cores as 4"); |
| sysCpuNum = 4; |
| } |
| int cpuPortNum = (int) (sysCpuNum / cpuWeight); |
| if (cpuPortNum < 1) { |
| |
| LOG.info("Invalid supervisor.slots.port.cpu.weight setting :" + cpuWeight + ", cpu cores:" + sysCpuNum); |
| cpuPortNum = 1; |
| } |
| |
| Double memWeight = ConfigExtension.getSupervisorSlotsPortMemWeight(conf); |
| int memPortNum = Integer.MAX_VALUE; |
| Long physicalMemSize = JStormUtils.getPhysicMemorySize(); |
| if (physicalMemSize == null) { |
| LOG.info("Failed to get memory size"); |
| } else { |
| LOG.info("Get system memory size :" + physicalMemSize); |
| long workerMemSize = ConfigExtension.getMemSizePerWorker(conf); |
| memPortNum = (int) (physicalMemSize / (workerMemSize * memWeight)); |
| if (memPortNum < 1) { |
| LOG.info("Invalide worker.memory.size setting:" + workerMemSize); |
| memPortNum = 4; |
| } else if (memPortNum < 4) { |
| LOG.info("System memory is too small for jstorm"); |
| memPortNum = 4; |
| } |
| } |
| |
| int portNum = Math.min(cpuPortNum, memPortNum); |
| if (portNum < 1) { |
| portNum = 1; |
| } |
| |
| int portBase = ConfigExtension.getSupervisorSlotsPortsBase(conf); |
| portList = new ArrayList<Integer>(); |
| for (int i = 0; i < portNum; i++) { |
| portList.add(portBase + i); |
| } |
| |
| return portList; |
| } |
| |
| public static byte[] longToBytes(long x) { |
| ByteBuffer buffer = ByteBuffer.allocate(Long.SIZE); |
| buffer.putLong(x); |
| return buffer.array(); |
| } |
| |
| public static long bytesToLong(byte[] bytes) { |
| ByteBuffer buffer = ByteBuffer.allocate(Long.SIZE); |
| buffer.put(bytes); |
| buffer.flip();// need flip |
| return buffer.getLong(); |
| } |
| |
| public static Object createDisruptorWaitStrategy(Map conf) { |
| String waitStrategy = (String) conf.get(Config.TOPOLOGY_DISRUPTOR_WAIT_STRATEGY); |
| Object ret; |
| if (waitStrategy.contains("TimeoutBlockingWaitStrategy")) { |
| long timeout = parseLong(conf.get(Config.TOPOLOGY_DISRUPTOR_WAIT_TIMEOUT), 10); |
| ret = Utils.newInstance(waitStrategy, timeout, TimeUnit.MILLISECONDS); |
| } else { |
| ret = Utils.newInstance(waitStrategy); |
| } |
| |
| return ret; |
| } |
| |
| public static Object thriftToObject(Object obj) { |
| Object ret = null; |
| if (obj instanceof org.apache.thrift.TBase) { |
| ret = thriftToMap((org.apache.thrift.TBase)obj); |
| }else if (obj instanceof List) { |
| ret = new ArrayList<>(); |
| for (Object item : (List)obj) { |
| ((List)ret).add(thriftToObject(item)); |
| } |
| }else if (obj instanceof Map) { |
| ret = new HashMap<String, Object>(); |
| Set<Entry> entrySet = ((Map)obj).entrySet(); |
| for (Entry entry : entrySet) { |
| ((Map)ret).put(String.valueOf(entry.getKey()), thriftToObject(entry.getValue())); |
| } |
| }else { |
| |
| ret = String.valueOf(obj); |
| } |
| |
| return ret; |
| } |
| |
| public static Map<String, Object> thriftToMap( |
| org.apache.thrift.TBase thriftObj) { |
| Map<String, Object> ret = new HashMap<String, Object>(); |
| |
| int i = 1; |
| TFieldIdEnum field = thriftObj.fieldForId(i); |
| while(field != null) { |
| if (thriftObj.isSet(field)) { |
| Object obj = thriftObj.getFieldValue(field); |
| ret.put(field.getFieldName(), thriftToObject(obj)); |
| |
| } |
| field = thriftObj.fieldForId(++i); |
| } |
| |
| return ret; |
| } |
| |
| public static List<Map<String, Object>> thriftToMap(List thriftObjs) { |
| List<Map<String, Object> > ret = new ArrayList<Map<String, Object> > () ; |
| |
| for (Object thriftObj : thriftObjs) { |
| ret.add(thriftToMap((org.apache.thrift.TBase)thriftObj)); |
| } |
| |
| return ret; |
| } |
| |
| |
| public static long halfValueOfSum(long v1, long v2, boolean increment) { |
| long ret = (v1 + v2) / 2; |
| if (increment) { |
| ret += (v1 + v2) % 2; |
| } |
| return ret; |
| } |
| } |
