[Improvement][api] https://github.com/datavane/datasophon/issues/364 Improvement

datasophon-api/src/main/resources/assembly.xml

@@ -93,6 +93,19 @@
             <!-- Linux 脚本，需要打包为 unix 格式，否则换行可能有问题 -->
             <lineEnding>unix</lineEnding>
         </fileSet>
+        <!--拷贝脚本文件到scripts目录下面-->
+        <fileSet>
+            <directory>${basedir}/../scripts</directory>
+            <includes>
+                <include>*.sh</include>
+            </includes>
+            <fileMode>0755</fileMode>
+            <directoryMode>0755</directoryMode>
+            <filtered>true</filtered>
+            <outputDirectory>scripts</outputDirectory>
+            <!-- Linux 脚本，需要打包为 unix 格式，否则换行可能有问题 -->
+            <lineEnding>unix</lineEnding>
+        </fileSet>
 
 
         <fileSet>

datasophon-common/src/main/java/com/datasophon/common/Constants.java

@@ -47,6 +47,7 @@ public final class Constants {
     public static final String WORKER_PACKAGE_NAME = "datasophon-worker.tar.gz";
     public static final String WORKER_SCRIPT_PATH = INSTALL_PATH + "/datasophon-worker/script/";
     public static final String WORKER_PATH = INSTALL_PATH + "/datasophon-worker";
+    public static final String SHELL_SCRIPT_PATH = "/scripts";
 
     public static final String CLUSTER_ID = "cluster_id";
     public static final String MANAGED = "managed";
@@ -255,4 +256,6 @@ private Constants() {
      * root user
      */
     public static final String ROOT = "root";
+
+    public static final String DISPATCHER_WORK = "dispatcher-worker.sh";
 }

datasophon-common/src/main/java/com/datasophon/common/utils/HostUtils.java

@@ -84,4 +84,22 @@ public static String getIp(String hostName) {
         }
     }
 
+    public static String getLocalIp() {
+        try {
+            InetAddress ip = InetAddress.getLocalHost();
+            return ip.getHostAddress();
+        } catch (UnknownHostException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public static String getLocalHostName() {
+        try {
+            InetAddress ip = InetAddress.getLocalHost();
+            return ip.getHostName();
+        } catch (UnknownHostException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
 }

datasophon-service/src/main/java/com/datasophon/api/master/DispatcherWorkerActor.java

@@ -19,17 +19,23 @@
 
 package com.datasophon.api.master;
 
+import cn.hutool.core.io.FileUtil;
 import com.datasophon.api.master.handler.host.CheckWorkerMd5Handler;
 import com.datasophon.api.master.handler.host.DecompressWorkerHandler;
 import com.datasophon.api.master.handler.host.DispatcherWorkerHandlerChain;
 import com.datasophon.api.master.handler.host.InstallJDKHandler;
 import com.datasophon.api.master.handler.host.StartWorkerHandler;
 import com.datasophon.api.master.handler.host.UploadWorkerHandler;
+import com.datasophon.api.utils.CommonUtils;
+import com.datasophon.api.utils.ExecuteShellScriptUtils;
 import com.datasophon.api.utils.MessageResolverUtils;
 import com.datasophon.api.utils.MinaUtils;
+import com.datasophon.common.Constants;
 import com.datasophon.common.command.DispatcherHostAgentCommand;
+import com.datasophon.common.enums.InstallState;
 import com.datasophon.common.model.HostInfo;
 
+import com.datasophon.common.utils.HostUtils;
 import org.apache.sshd.client.session.ClientSession;
 
 import scala.Option;
@@ -40,6 +46,8 @@
 import akka.actor.UntypedActor;
 import cn.hutool.core.util.ObjectUtil;
 
+import java.nio.charset.Charset;
+
 public class DispatcherWorkerActor extends UntypedActor {
 
     private static final Logger logger = LoggerFactory.getLogger(DispatcherWorkerActor.class);
@@ -54,6 +62,8 @@ public void preRestart(Throwable reason, Option<Object> message) throws Exceptio
     public void onReceive(Object message) throws Throwable {
         DispatcherHostAgentCommand command = (DispatcherHostAgentCommand) message;
         HostInfo hostInfo = command.getHostInfo();
+        String localIp = HostUtils.getLocalIp();
+        String localHostName = HostUtils.getLocalHostName();
         logger.info("start dispatcher host agent :{}", hostInfo.getHostname());
         hostInfo.setMessage(
                 MessageResolverUtils.getMessage(
@@ -62,9 +72,37 @@ public void onReceive(Object message) throws Throwable {
                 MinaUtils.openConnection(
                         hostInfo.getHostname(), hostInfo.getSshPort(), hostInfo.getSshUser());
         DispatcherWorkerHandlerChain handlerChain = new DispatcherWorkerHandlerChain();
-        handlerChain.addHandler(new UploadWorkerHandler());
-        handlerChain.addHandler(new CheckWorkerMd5Handler());
-        handlerChain.addHandler(new DecompressWorkerHandler());
+        if(hostInfo.getIp().equals(localIp)){
+            String currDir = System.getProperty("user.dir");
+            String executeShellPath = currDir+ Constants.SHELL_SCRIPT_PATH + Constants.SLASH+Constants.DISPATCHER_WORK;
+            logger.info("start dispatcher manage node host agent,execute shell path :{}", executeShellPath);
+            String md5 = FileUtil.readString(
+                    Constants.MASTER_MANAGE_PACKAGE_PATH +
+                            Constants.SLASH +
+                            Constants.WORKER_PACKAGE_NAME + ".md5", Charset.defaultCharset()).trim();
+            int exeCode = ExecuteShellScriptUtils.executeShellScript(executeShellPath,md5);
+            if(0 == exeCode){
+                logger.info("distribution  datasophon-worker.tar.gz success");
+                logger.info("md5.verification datasophon-worker.tar.gz success");
+                logger.info("decompress datasophon-worker.tar.gz success");
+                hostInfo.setProgress(50);
+                hostInfo.setMessage(MessageResolverUtils
+                        .getMessage("installation.package.decompressed.success.and.modify.configuration.file"));
+            }else {
+                logger.error("dispatcher manage node host agent failed");
+                hostInfo.setErrMsg("dispatcher manage node host agent failed");
+                hostInfo.setMessage(MessageResolverUtils
+                        .getMessage("dispatcher manage node host agent failed"));
+                CommonUtils.updateInstallState(InstallState.FAILED, hostInfo);
+                throw new RuntimeException("---- dispatcher manage node host agent failed ----");
+            }
+
+        }else {
+            handlerChain.addHandler(new UploadWorkerHandler());
+            handlerChain.addHandler(new CheckWorkerMd5Handler());
+            handlerChain.addHandler(new DecompressWorkerHandler());
+        }
+
         handlerChain.addHandler(new InstallJDKHandler());
         handlerChain.addHandler(
                 new StartWorkerHandler(command.getClusterId(), command.getClusterFrame()));

datasophon-service/src/main/java/com/datasophon/api/master/HostConnectActor.java

@@ -25,6 +25,7 @@
 import com.datasophon.common.model.CheckResult;
 import com.datasophon.common.model.HostInfo;
 
+import com.datasophon.common.utils.HostUtils;
 import org.apache.sshd.client.session.ClientSession;
 
 import scala.Option;
@@ -48,22 +49,33 @@ public void preRestart(Throwable reason, Option<Object> message) throws Exceptio
     @Override
     public void onReceive(Object message) throws Throwable {
         if (message instanceof HostCheckCommand) {
+            String localIp = HostUtils.getLocalIp();
+            String localHostName = HostUtils.getLocalHostName();
+            logger.info("datasophon manager install hostname and ip :"+localHostName+",",localIp);
             HostCheckCommand hostCheckCommand = (HostCheckCommand) message;
             HostInfo hostInfo = hostCheckCommand.getHostInfo();
             logger.info("start host check:{}", hostInfo.getHostname());
-            ClientSession session =
-                    MinaUtils.openConnection(
-                            hostInfo.getHostname(), hostInfo.getSshPort(), hostInfo.getSshUser());
-            if (ObjectUtil.isNotNull(session)) {
+            if(hostInfo.getIp().equals(localIp)){
+                logger.info("datasophon manager node doesn't need to be checked");
                 hostInfo.setCheckResult(
                         new CheckResult(
                                 Status.CHECK_HOST_SUCCESS.getCode(),
                                 Status.CHECK_HOST_SUCCESS.getMsg()));
-            } else {
-                hostInfo.setCheckResult(
-                        new CheckResult(
-                                Status.CONNECTION_FAILED.getCode(),
-                                Status.CONNECTION_FAILED.getMsg()));
+            }else {
+                ClientSession session =
+                        MinaUtils.openConnection(
+                                hostInfo.getHostname(), hostInfo.getSshPort(), hostInfo.getSshUser());
+                if (ObjectUtil.isNotNull(session)) {
+                    hostInfo.setCheckResult(
+                            new CheckResult(
+                                    Status.CHECK_HOST_SUCCESS.getCode(),
+                                    Status.CHECK_HOST_SUCCESS.getMsg()));
+                } else {
+                    hostInfo.setCheckResult(
+                            new CheckResult(
+                                    Status.CONNECTION_FAILED.getCode(),
+                                    Status.CONNECTION_FAILED.getMsg()));
+                }
                 MinaUtils.closeConnection(session);
             }
             logger.info("end host check:{}", hostInfo.getHostname());

datasophon-service/src/main/java/com/datasophon/api/utils/ExecuteShellScriptUtils.java

@@ -0,0 +1,25 @@
+package com.datasophon.api.utils;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+
+import com.datasophon.api.master.DispatcherWorkerActor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ExecuteShellScriptUtils {
+    private static final Logger logger = LoggerFactory.getLogger(ExecuteShellScriptUtils.class);
+    public static int executeShellScript(String shellScriptPath,String params) throws IOException, InterruptedException {
+        logger.info("start executeShellScript:{}", shellScriptPath);
+        ProcessBuilder pb = new ProcessBuilder("sh",shellScriptPath,params);
+        //pb.inheritIO();
+        Process p = pb.start();
+        /*BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()));
+        String line;
+        while ((line = reader.readLine()) != null){
+            System.out.println(line);
+        }*/
+        return p.waitFor();
+    }
+}

scripts/dispatcher-worker.sh

@@ -0,0 +1,30 @@
+#!/bin/bash
+
+# example: sh init-jdk.sh
+# instal and config jdk env
+workmd5="$1"
+if [ $UID -ne 0 ]; then
+  echo Non root user. Please run as root.
+  exit 1
+fi
+if [ -L $0 ]; then
+  BASE_DIR=$(dirname $(readlink $0))
+else
+  BASE_DIR=$(dirname $0)
+fi
+BASE_PATH=$(
+  cd ${BASE_DIR}
+  pwd
+)
+INIT_PATH=$(dirname "${BASE_PATH}")
+echo "INIT_PATH: ${INIT_PATH}"
+
+cp /opt/datasophon/DDP/packages/datasophon-worker.tar.gz /opt/datasophon/
+checkworkmd5=$(md5sum /opt/datasophon/DDP/packages/datasophon-worker.tar.gz | awk '{print $1}')
+if [ "$checkworkmd5" = "$workmd5" ]; then
+  echo "md5效验通过"
+else
+  echo "md5效验不通过"
+  exit 1
+fi
+tar -zxvf /opt/datasophon/datasophon-worker.tar.gz -C /opt/datasophon/