Merge branch 'dev' of https://github.com/datasophon/datasophon into dev

datasophon · datasophon · commit 0f136861b2a7 · 2023-04-24T23:35:44.000+08:00
diff --git a/datasophon-api/src/main/java/com/datasophon/api/master/ActorUtils.java b/datasophon-api/src/main/java/com/datasophon/api/master/ActorUtils.java
@@ -67,6 +67,8 @@ public static void init() throws UnknownHostException, NoSuchAlgorithmException
                 getActorRefName(ServiceRoleCheckActor.class));
         ActorRef hostCheckActor =
                 actorSystem.actorOf(Props.create(HostCheckActor.class), getActorRefName(HostCheckActor.class));
+        actorSystem.actorOf(Props.create(MasterNodeProcessingActor.class),
+                getActorRefName(MasterNodeProcessingActor.class));
 
         actorSystem.scheduler().schedule(
                 FiniteDuration.apply(60L, TimeUnit.SECONDS),
diff --git a/datasophon-api/src/main/java/com/datasophon/api/master/MasterNodeProcessingActor.java b/datasophon-api/src/main/java/com/datasophon/api/master/MasterNodeProcessingActor.java
@@ -0,0 +1,52 @@
+package com.datasophon.api.master;
+
+import akka.actor.UntypedActor;
+import com.datasophon.common.command.OlapOpsType;
+import com.datasophon.common.command.OlapSqlExecCommand;
+import com.datasophon.common.utils.ExecResult;
+import com.datasophon.common.utils.OlapUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.TimeUnit;
+
+public class MasterNodeProcessingActor extends UntypedActor {
+
+    private static final Logger logger = LoggerFactory.getLogger(MasterNodeProcessingActor.class);
+
+    @Override
+    public void onReceive(Object message) throws Throwable {
+        if (message instanceof OlapSqlExecCommand) {
+            OlapSqlExecCommand command = (OlapSqlExecCommand) message;
+            ExecResult execResult = OlapOpsType.ADD_BE.equals(command.getOpsType())
+                    ? OlapUtils.addBackendBySqlClient(command.getFeMaster(), command.getHostName())
+                    : OlapUtils.addFollowerBySqlClient(command.getFeMaster(), command.getHostName());
+            String tip = OlapOpsType.ADD_BE.equals(command.getOpsType()) ? "backend" : "follower";
+            if (execResult.getExecResult()) {
+                logger.info(command.getHostName() + " " + tip + " be added success");
+            } else {
+                logger.info(command.getHostName() + " " + tip + " be added failed");
+            }
+            int tryTimes = 0;
+            while (!execResult.getExecResult() && tryTimes < 3) {
+                try {
+                    TimeUnit.SECONDS.sleep(10L);
+                    execResult = OlapOpsType.ADD_BE.equals(command.getOpsType())
+                            ? OlapUtils.addBackendBySqlClient(command.getFeMaster(), command.getHostName())
+                            : OlapUtils.addFollowerBySqlClient(command.getFeMaster(), command.getHostName());
+                    if (execResult.getExecResult()) {
+                        logger.info(command.getHostName() + " " + tip + " be added success");
+                        break;
+                    } else {
+                        logger.info(command.getHostName() + " " + tip + " be added failed");
+                    }
+                    tryTimes++;
+                } catch (InterruptedException e) {
+                    logger.info("The SR operate be sleep operation failed");
+                }
+            }
+        } else {
+            unhandled(message);
+        }
+    }
+}
diff --git a/datasophon-api/src/main/java/com/datasophon/api/master/handler/service/ServiceStartHandler.java b/datasophon-api/src/main/java/com/datasophon/api/master/handler/service/ServiceStartHandler.java
@@ -19,6 +19,8 @@
 
 import com.datasophon.api.load.GlobalVariables;
 import com.datasophon.api.master.ActorUtils;
+import com.datasophon.common.Constants;
+import com.datasophon.common.cache.CacheUtils;
 import com.datasophon.common.command.ServiceRoleOperateCommand;
 import com.datasophon.common.enums.ServiceRoleType;
 import com.datasophon.common.model.ServiceRoleInfo;
@@ -56,6 +58,7 @@ public ExecResult handlerRequest(ServiceRoleInfo serviceRoleInfo) throws Excepti
         serviceRoleOperateCommand.setSlave(serviceRoleInfo.isSlave());
         serviceRoleOperateCommand.setCommandType(serviceRoleInfo.getCommandType());
         serviceRoleOperateCommand.setMasterHost(serviceRoleInfo.getMasterHost());
+        serviceRoleOperateCommand.setManagerHost(CacheUtils.getString(Constants.HOSTNAME));
 
         logger.info("service master host is {}", serviceRoleInfo.getMasterHost());
 
diff --git a/datasophon-common/src/main/java/com/datasophon/common/command/OlapOpsType.java b/datasophon-common/src/main/java/com/datasophon/common/command/OlapOpsType.java
@@ -0,0 +1,13 @@
+package com.datasophon.common.command;
+
+/**
+ * @Enum OlapOpsType
+ * @Author: 张大伟
+ * @Date: 2023/4/24 21:28
+ * @Version: 1.0
+ */
+public enum OlapOpsType {
+
+    ADD_BE,
+    ADD_FE
+}
diff --git a/datasophon-common/src/main/java/com/datasophon/common/command/OlapSqlExecCommand.java b/datasophon-common/src/main/java/com/datasophon/common/command/OlapSqlExecCommand.java
@@ -0,0 +1,15 @@
+package com.datasophon.common.command;
+
+import java.io.Serializable;
+
+import lombok.Data;
+
+@Data
+public class OlapSqlExecCommand implements Serializable {
+
+    private OlapOpsType opsType;
+
+    private String feMaster;
+
+    private String hostName;
+}
diff --git a/datasophon-common/src/main/java/com/datasophon/common/command/ServiceRoleOperateCommand.java b/datasophon-common/src/main/java/com/datasophon/common/command/ServiceRoleOperateCommand.java
@@ -40,6 +40,8 @@ public class ServiceRoleOperateCommand extends BaseCommand implements Serializab
 
     private String masterHost;
 
+    private String managerHost;
+
     private Boolean enableRangerPlugin;
 
     private RunAs runAs;
diff --git a/datasophon-common/src/main/java/com/datasophon/common/utils/OlapUtils.java b/datasophon-common/src/main/java/com/datasophon/common/utils/OlapUtils.java
@@ -70,7 +70,7 @@ public static ExecResult addFollowerBySqlClient(String feMaster,
         return ShellUtils.exceShell(sqlCommand);
     }
 
-    public static ExecResult addBackendSqlClient(String feMaster,
+    public static ExecResult addBackendBySqlClient(String feMaster,
                                                  String hostname) {
         String sqlCommand =
                 "mysql -h"
diff --git a/datasophon-worker/src/main/java/com/datasophon/worker/WorkerApplicationServer.java b/datasophon-worker/src/main/java/com/datasophon/worker/WorkerApplicationServer.java
@@ -28,6 +28,7 @@
 import com.datasophon.common.utils.ShellUtils;
 import com.datasophon.worker.actor.RemoteEventActor;
 import com.datasophon.worker.actor.WorkerActor;
+import com.datasophon.worker.utils.ActorUtils;
 import com.datasophon.worker.utils.UnixUtils;
 
 import java.net.InetAddress;
@@ -77,6 +78,7 @@ public static void main(String[] args) throws UnknownHostException {
         CacheUtils.put(Constants.HOSTNAME, hostname);
         // init actor
         ActorSystem system = initActor(hostname);
+        ActorUtils.setActorSystem(system);
 
         subscribeRemoteEvent(system);
 
diff --git a/datasophon-worker/src/main/java/com/datasophon/worker/strategy/BEHandlerStrategy.java b/datasophon-worker/src/main/java/com/datasophon/worker/strategy/BEHandlerStrategy.java
@@ -17,15 +17,19 @@
 
 package com.datasophon.worker.strategy;
 
+import akka.actor.ActorRef;
 import com.datasophon.common.Constants;
 import com.datasophon.common.cache.CacheUtils;
+import com.datasophon.common.command.OlapOpsType;
+import com.datasophon.common.command.OlapSqlExecCommand;
 import com.datasophon.common.command.ServiceRoleOperateCommand;
 import com.datasophon.common.enums.CommandType;
 import com.datasophon.common.utils.ExecResult;
 import com.datasophon.common.utils.OlapUtils;
 import com.datasophon.common.utils.ShellUtils;
 import com.datasophon.common.utils.ThrowableUtils;
 import com.datasophon.worker.handler.ServiceHandler;
+import com.datasophon.worker.utils.ActorUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -48,8 +52,13 @@ public ExecResult handler(ServiceRoleOperateCommand command) {
                     command.getDecompressPackageName(), command.getRunAs());
             if (startResult.getExecResult()) {
                 try {
-                    OlapUtils.addBackend(command.getMasterHost(), CacheUtils.getString(Constants.HOSTNAME));
-                } catch (SQLException | ClassNotFoundException e) {
+                    OlapSqlExecCommand sqlExecCommand = new OlapSqlExecCommand();
+                    sqlExecCommand.setFeMaster(command.getMasterHost());
+                    sqlExecCommand.setHostName(CacheUtils.getString(Constants.HOSTNAME));
+                    sqlExecCommand.setOpsType(OlapOpsType.ADD_BE);
+                    ActorUtils.getRemoteActor(command.getManagerHost(), "masterNodeProcessingActor")
+                            .tell(sqlExecCommand, ActorRef.noSender());
+                } catch (Exception e) {
                     logger.info("add backend failed {}", ThrowableUtils.getStackTrace(e));
                 }
                 logger.info("slave be start success");
diff --git a/datasophon-worker/src/main/java/com/datasophon/worker/strategy/FEHandlerStrategy.java b/datasophon-worker/src/main/java/com/datasophon/worker/strategy/FEHandlerStrategy.java
@@ -17,8 +17,11 @@
 
 package com.datasophon.worker.strategy;
 
+import akka.actor.ActorRef;
 import com.datasophon.common.Constants;
 import com.datasophon.common.cache.CacheUtils;
+import com.datasophon.common.command.OlapOpsType;
+import com.datasophon.common.command.OlapSqlExecCommand;
 import com.datasophon.common.command.ServiceRoleOperateCommand;
 import com.datasophon.common.enums.CommandType;
 import com.datasophon.common.model.ServiceRoleRunner;
@@ -30,6 +33,7 @@
 import java.sql.SQLException;
 import java.util.ArrayList;
 
+import com.datasophon.worker.utils.ActorUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -58,8 +62,13 @@ public ExecResult handler(ServiceRoleOperateCommand command) {
                 if (startResult.getExecResult()) {
                     // add follower
                     try {
-                        OlapUtils.addFollower(command.getMasterHost(), CacheUtils.getString(Constants.HOSTNAME));
-                    } catch (SQLException | ClassNotFoundException e) {
+                        OlapSqlExecCommand sqlExecCommand = new OlapSqlExecCommand();
+                        sqlExecCommand.setFeMaster(command.getMasterHost());
+                        sqlExecCommand.setHostName(CacheUtils.getString(Constants.HOSTNAME));
+                        sqlExecCommand.setOpsType(OlapOpsType.ADD_FE);
+                        ActorUtils.getRemoteActor(command.getManagerHost(), "masterNodeProcessingActor")
+                                .tell(sqlExecCommand, ActorRef.noSender());
+                    } catch (Exception e) {
                         logger.info("add slave fe failed {}", ThrowableUtils.getStackTrace(e));
                     }
                     logger.info("slave fe start success");
diff --git a/datasophon-worker/src/main/java/com/datasophon/worker/utils/ActorUtils.java b/datasophon-worker/src/main/java/com/datasophon/worker/utils/ActorUtils.java
@@ -0,0 +1,36 @@
+package com.datasophon.worker.utils;
+
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+
+import java.util.concurrent.TimeUnit;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSelection;
+import akka.actor.ActorSystem;
+import akka.util.Timeout;
+
+public class ActorUtils {
+
+    private static ActorSystem actorSystem;
+
+    public static void setActorSystem(ActorSystem actorSystem) {
+        ActorUtils.actorSystem = actorSystem;
+    }
+
+    public static ActorRef getRemoteActor(String hostname, String actorName) {
+        String actorPath = "akka.tcp://datasophon@" + hostname + ":2551/user/" + actorName;
+        ActorSelection actorSelection = actorSystem.actorSelection(actorPath);
+        Timeout timeout = new Timeout(Duration.create(30, TimeUnit.SECONDS));
+        Future<ActorRef> future = actorSelection.resolveOne(timeout);
+        ActorRef actorRef = null;
+        try {
+            actorRef = Await.result(future, Duration.create(30, TimeUnit.SECONDS));
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+        return actorRef;
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -70,7 +70,7 @@ public static ExecResult addFollowerBySqlClient(String feMaster,`
`70`	`70`	`return ShellUtils.exceShell(sqlCommand);`
`71`	`71`	`}`
`72`	`72`
`73`		`- public static ExecResult addBackendSqlClient(String feMaster,`
	`73`	`+ public static ExecResult addBackendBySqlClient(String feMaster,`
`74`	`74`	`String hostname) {`
`75`	`75`	`String sqlCommand =`
`76`	`76`	`"mysql -h"`