Merge branch 'master' of https://github.com/fddc/PowerJob into 4.2.0-win-support

powerjob-official-processors/src/main/java/tech/powerjob/official/processors/impl/script/AbstractScriptProcessor.java

@@ -13,6 +13,7 @@ import tech.powerjob.official.processors.util.CommonUtils;
 
 import java.io.*;
 import java.net.URL;
+import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.util.Set;
 import java.util.concurrent.ForkJoinPool;
@@ -30,6 +31,7 @@ public abstract class AbstractScriptProcessor extends CommonBasicProcessor {
     private static final ForkJoinPool POOL = new ForkJoinPool(4 * Runtime.getRuntime().availableProcessors());
     private static final Set<String> DOWNLOAD_PROTOCOL = Sets.newHashSet("http", "https", "ftp");
     protected static final String SH_SHELL = "/bin/sh";
+    protected static final String CMD_SHELL = "cmd.exe";
 
     private static final String WORKER_DIR = System.getProperty("user.home") + "/powerjob/worker/official_script_processor/";
 
@@ -55,13 +57,16 @@ public abstract class AbstractScriptProcessor extends CommonBasicProcessor {
         }
 
         // 授权
-        ProcessBuilder chmodPb = new ProcessBuilder("/bin/chmod", "755", scriptPath);
+        if  ( !SystemUtils.IS_OS_WINDOWS) {
-        // 等待返回，这里不可能导致死锁（shell产生大量数据可能导致死锁）
+            ProcessBuilder chmodPb = new ProcessBuilder("/bin/chmod", "755", scriptPath);
-        chmodPb.start().waitFor();
+            // 等待返回，这里不可能导致死锁（shell产生大量数据可能导致死锁）
-        omsLogger.info("[SYSTEM] chmod 755 authorization complete, ready to start execution~");
+            chmodPb.start().waitFor();
-
+            omsLogger.info("[SYSTEM] chmod 755 authorization complete, ready to start execution~");
+        }
         // 2. 执行目标脚本
-        ProcessBuilder pb = new ProcessBuilder(getRunCommand(), scriptPath);
+        ProcessBuilder pb = StringUtils.equals(getRunCommand(), CMD_SHELL) ?
+                new ProcessBuilder(getRunCommand(), "/c", scriptPath)
+                : new ProcessBuilder(getRunCommand(), scriptPath);
         Process process = pb.start();
 
         StringBuilder inputBuilder = new StringBuilder();
@@ -70,10 +75,12 @@ public abstract class AbstractScriptProcessor extends CommonBasicProcessor {
         boolean success = true;
         String result;
 
+        //解决windows平台中文乱码
+        Charset loggerCharset =  SystemUtils.IS_OS_WINDOWS? Charset.forName("GBK"):StandardCharsets.UTF_8;
         try (InputStream is = process.getInputStream(); InputStream es = process.getErrorStream()) {
 
-            POOL.execute(() -> copyStream(is, inputBuilder, omsLogger));
+            POOL.execute(() -> copyStream(is, inputBuilder, omsLogger,loggerCharset));
-            POOL.execute(() -> copyStream(es, errorBuilder, omsLogger));
+            POOL.execute(() -> copyStream(es, errorBuilder, omsLogger,loggerCharset));
 
             success = process.waitFor() == 0;
 
@@ -107,16 +114,25 @@ public abstract class AbstractScriptProcessor extends CommonBasicProcessor {
         }
 
         // 非下载链接，为 processInfo 生成可执行文件
-        try (FileWriter fw = new FileWriter(script); BufferedWriter bw = new BufferedWriter(fw)) {
+        if(SystemUtils.IS_OS_WINDOWS)
-            bw.write(processorInfo);
+        {
-            bw.flush();
+            try (Writer fstream = new OutputStreamWriter(new FileOutputStream(script), Charset.forName("GBK")); BufferedWriter out = new BufferedWriter(fstream)) {
+                out.write(processorInfo);
+                out.flush();
+            }
+        }
+        else {
+            try (FileWriter fw = new FileWriter(script); BufferedWriter bw = new BufferedWriter(fw)) {
+                bw.write(processorInfo);
+                bw.flush();
+            }
         }
         return scriptPath;
     }
 
-    private static void copyStream(InputStream is, StringBuilder sb, OmsLogger omsLogger) {
+    private static void copyStream(InputStream is, StringBuilder sb, OmsLogger omsLogger, Charset charset) {
         String line;
-        try (BufferedReader br = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) {
+        try (BufferedReader br = new BufferedReader(new InputStreamReader(is, charset))) {
             while ((line = br.readLine()) != null) {
                 sb.append(line);
                 // 同步到在线日志

powerjob-official-processors/src/main/java/tech/powerjob/official/processors/impl/script/CMDProcessor.java

@@ -0,0 +1,20 @@
+package tech.powerjob.official.processors.impl.script;
+
+/**
+ * python processor
+ *
+ * @author fddc
+ * @since 2021/5/14
+ */
+public class CMDProcessor extends AbstractScriptProcessor {
+
+    @Override
+    protected String getScriptName(Long instanceId) {
+        return String.format("cmd_%d.bat", instanceId);
+    }
+
+    @Override
+    protected String getRunCommand() {
+        return "cmd.exe";
+    }
+}

powerjob-official-processors/src/main/java/tech/powerjob/official/processors/impl/script/PowerShellProcessor.java

@@ -0,0 +1,20 @@
+package tech.powerjob.official.processors.impl.script;
+
+/**
+ * python processor
+ *
+ * @author fddc
+ * @since 2021/5/14
+ */
+public class PowerShellProcessor extends AbstractScriptProcessor {
+
+    @Override
+    protected String getScriptName(Long instanceId) {
+        return String.format("powershell_%d.ps1", instanceId);
+    }
+
+    @Override
+    protected String getRunCommand() {
+        return "powershell.exe";
+    }
+}

powerjob-worker-agent/src/main/java/tech/powerjob/agent/MainApplication.java

@@ -36,6 +36,9 @@ public class MainApplication implements Runnable {
     @Option(names = {"-l", "--length"}, description = "ProcessResult#msg max length")
     private int length = 1024;
 
+    @Option(names = {"-t", "--tag"}, description = "worker-agent's tag")
+    private String tag;
+
     public static void main(String[] args) {
         CommandLine commandLine = new CommandLine(new MainApplication());
         commandLine.execute(args);
@@ -52,6 +55,7 @@ public class MainApplication implements Runnable {
             cfg.setServerAddress(Splitter.on(",").splitToList(server));
             cfg.setStoreStrategy(StoreStrategy.MEMORY.name().equals(storeStrategy) ? StoreStrategy.MEMORY : StoreStrategy.DISK);
             cfg.setMaxResultLength(length);
+            cfg.setTag(tag);
 
             PowerJobWorker worker = new PowerJobWorker();
             worker.setConfig(cfg);