Add nebula engine to linkis

ChengJie1053 · Sep 13, 2023 · 89157b8 · 89157b8
1 parent 4dc6e25
commit 89157b8
Show file tree

Hide file tree

Showing 13 changed files with 161 additions and 290 deletions.
diff --git a/...e/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/LinkisSQLConnection.scala b/...e/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/LinkisSQLConnection.scala
@@ -431,6 +431,7 @@ class LinkisSQLConnection(private[jdbc] val ujesClient: UJESClient, props: Prope
       case EngineType.HIVE => RunType.HIVE
       case EngineType.TRINO => RunType.TRINO_SQL
       case EngineType.PRESTO => RunType.PRESTO_SQL
+      case EngineType.NEBULA => RunType.NEBULA_SQL
       case EngineType.ELASTICSEARCH => RunType.ES_SQL
       case EngineType.JDBC => RunType.JDBC
       case EngineType.PYTHON => RunType.SHELL

diff --git a/...-application-manager/src/main/java/org/apache/linkis/manager/am/conf/AMConfiguration.java b/...-application-manager/src/main/java/org/apache/linkis/manager/am/conf/AMConfiguration.java
@@ -68,7 +68,8 @@ public class AMConfiguration {
 
   public static final CommonVars<String> MULTI_USER_ENGINE_TYPES =
       CommonVars.apply(
-          "wds.linkis.multi.user.engine.types", "jdbc,es,presto,io_file,appconn,openlookeng,trino");
+          "wds.linkis.multi.user.engine.types",
+          "jdbc,es,presto,io_file,appconn,openlookeng,trino,nebula");
 
   public static final CommonVars<String> ALLOW_BATCH_KILL_ENGINE_TYPES =
       CommonVars.apply("wds.linkis.allow.batch.kill.engine.types", "spark,hive,python");
@@ -104,8 +105,8 @@ public class AMConfiguration {
   public static String getDefaultMultiEngineUser() {
     String jvmUser = Utils.getJvmUser();
     return String.format(
-        "{jdbc:\"%s\", es: \"%s\", presto:\"%s\", appconn:\"%s\", openlookeng:\"%s\", trino:\"%s\", io_file:\"root\"}",
-        jvmUser, jvmUser, jvmUser, jvmUser, jvmUser, jvmUser);
+        "{jdbc:\"%s\", es: \"%s\", presto:\"%s\", appconn:\"%s\", openlookeng:\"%s\", trino:\"%s\", nebula:\"%s\",io_file:\"root\"}",
+        jvmUser, jvmUser, jvmUser, jvmUser, jvmUser, jvmUser, jvmUser);
   }
 
   public static boolean isMultiUserEngine(String engineType) {

diff --git a/...is-label-common/src/main/java/org/apache/linkis/manager/label/conf/LabelCommonConfig.java b/...is-label-common/src/main/java/org/apache/linkis/manager/label/conf/LabelCommonConfig.java
@@ -69,6 +69,9 @@ public class LabelCommonConfig {
   public static final CommonVars<String> DATAX_ENGINE_VERSION =
       CommonVars.apply("wds.linkis.datax.engine.version", "3.0.0");
 
+  public static final CommonVars<String> NEBULA_ENGINE_VERSION =
+      CommonVars.apply("wds.linkis.nebula.engine.version", "3.0.0");
+
   public static final CommonVars<String> PRESTO_ENGINE_VERSION =
       CommonVars.apply("wds.linkis.presto.engine.version", "0.234");
 

diff --git a/...label-common/src/main/java/org/apache/linkis/manager/label/entity/engine/EngineType.scala b/...label-common/src/main/java/org/apache/linkis/manager/label/entity/engine/EngineType.scala
@@ -45,6 +45,8 @@ object EngineType extends Enumeration with Logging {
 
   val PRESTO = Value("presto")
 
+  val NEBULA = Value("nebula")
+
   val FLINK = Value("flink")
 
   val APPCONN = Value("appconn")
@@ -89,6 +91,7 @@ object EngineType extends Enumeration with Logging {
     case _ if IO_ENGINE_HDFS.toString.equalsIgnoreCase(str) => IO_ENGINE_HDFS
     case _ if PIPELINE.toString.equalsIgnoreCase(str) => PIPELINE
     case _ if PRESTO.toString.equalsIgnoreCase(str) => PRESTO
+    case _ if NEBULA.toString.equalsIgnoreCase(str) => NEBULA
     case _ if FLINK.toString.equalsIgnoreCase(str) => FLINK
     case _ if APPCONN.toString.equals(str) => APPCONN
     case _ if SQOOP.toString.equalsIgnoreCase(str) => SQOOP

diff --git a/...is-label-common/src/main/java/org/apache/linkis/manager/label/entity/engine/RunType.scala b/...is-label-common/src/main/java/org/apache/linkis/manager/label/entity/engine/RunType.scala
@@ -35,6 +35,7 @@ object RunType extends Enumeration {
   val PIPELINE = Value("pipeline")
   val JDBC = Value("jdbc")
   val PRESTO_SQL = Value("psql")
+  val NEBULA_SQL = Value("ngql")
   val JAR = Value("jar")
   val APPCONN = Value("appconn")
   val FUNCTION_MDQ_TYPE = Value("function.mdq")

diff --git a/...el-common/src/main/java/org/apache/linkis/manager/label/utils/EngineTypeLabelCreator.java b/...el-common/src/main/java/org/apache/linkis/manager/label/utils/EngineTypeLabelCreator.java
@@ -69,6 +69,8 @@ private static void init() {
               EngineType.FLINK().toString(), LabelCommonConfig.FLINK_ENGINE_VERSION.getValue());
           defaultVersion.put(
               EngineType.PRESTO().toString(), LabelCommonConfig.PRESTO_ENGINE_VERSION.getValue());
+          defaultVersion.put(
+              EngineType.NEBULA().toString(), LabelCommonConfig.NEBULA_ENGINE_VERSION.getValue());
           defaultVersion.put(
               EngineType.SQOOP().toString(), LabelCommonConfig.SQOOP_ENGINE_VERSION.getValue());
           defaultVersion.put(

diff --git a/linkis-engineconn-plugins/nebula/pom.xml b/linkis-engineconn-plugins/nebula/pom.xml
@@ -62,13 +62,12 @@
 
     <!-- nebula -->
     <dependency>
-        <groupId>com.vesoft</groupId>
-        <artifactId>client</artifactId>
-        <version>${nebula.version}</version>
+      <groupId>com.vesoft</groupId>
+      <artifactId>client</artifactId>
+      <version>${nebula.version}</version>
     </dependency>
   </dependencies>
 
-
   <build>
 
     <plugins>

diff --git a/...a/org/apache/linkis/engineplugin/nebula/builder/NebulaProcessEngineConnLaunchBuilder.java b/...a/org/apache/linkis/engineplugin/nebula/builder/NebulaProcessEngineConnLaunchBuilder.java
@@ -18,13 +18,5 @@
 package org.apache.linkis.engineplugin.nebula.builder;
 
 import org.apache.linkis.manager.engineplugin.common.launch.process.JavaProcessEngineConnLaunchBuilder;
-import org.apache.linkis.manager.label.entity.engine.UserCreatorLabel;
-import org.apache.linkis.storage.utils.StorageConfiguration;
 
-public class NebulaProcessEngineConnLaunchBuilder extends JavaProcessEngineConnLaunchBuilder {
-
-  @Override
-  public String getEngineStartUser(UserCreatorLabel label) {
-    return StorageConfiguration.HDFS_ROOT_USER.getValue();
-  }
-}
+public class NebulaProcessEngineConnLaunchBuilder extends JavaProcessEngineConnLaunchBuilder {}
diff --git a/.../nebula/src/main/java/org/apache/linkis/engineplugin/nebula/conf/NebulaConfiguration.java b/.../nebula/src/main/java/org/apache/linkis/engineplugin/nebula/conf/NebulaConfiguration.java
@@ -24,40 +24,27 @@ public class NebulaConfiguration {
   public static final CommonVars<Integer> ENGINE_CONCURRENT_LIMIT =
       CommonVars.apply("wds.linkis.engineconn.concurrent.limit", 100);
 
-  // unit in seconds
-  public static final CommonVars<Long> PRESTO_HTTP_CONNECT_TIME_OUT =
-      CommonVars.apply("wds.linkis.presto.http.connectTimeout", 60L);
-
-  public static final CommonVars<Long> PRESTO_HTTP_READ_TIME_OUT =
-      CommonVars.apply("wds.linkis.presto.http.readTimeout", 60L);
-
   public static final CommonVars<Integer> ENGINE_DEFAULT_LIMIT =
-      CommonVars.apply("wds.linkis.presto.default.limit", 5000);
-
-  public static final CommonVars<String> PRESTO_URL =
-      CommonVars.apply("wds.linkis.presto.url", "http://127.0.0.1:8080");
-
-  public static final CommonVars<String> PRESTO_RESOURCE_CONFIG_PATH =
-      CommonVars.apply("wds.linkis.presto.resource.config", "");
-
-  public static final CommonVars<String> PRESTO_USER_NAME =
-      CommonVars.apply("wds.linkis.presto.username", "default");
+      CommonVars.apply("wds.linkis.nebula.default.limit", 5000);
 
-  public static final CommonVars<String> PRESTO_PASSWORD =
-      CommonVars.apply("wds.linkis.presto.password", "");
+  public static final CommonVars<String> NEBULA_HOST =
+      CommonVars.apply("wds.linkis.nebula.host", "127.0.0.1");
 
-  public static final CommonVars<String> PRESTO_CATALOG =
-      CommonVars.apply("wds.linkis.presto.catalog", "system");
+  public static final CommonVars<Integer> NEBULA_PORT =
+      CommonVars.apply("wds.linkis.nebula.port", 9669);
 
-  public static final CommonVars<String> PRESTO_SCHEMA =
-      CommonVars.apply("wds.linkis.presto.schema", "");
+  public static final CommonVars<Integer> NEBULA_MAX_CONN_SIZE =
+      CommonVars.apply("wds.linkis.nebula.max.conn.size", 100);
 
-  public static final CommonVars<String> PRESTO_SOURCE =
-      CommonVars.apply("wds.linkis.presto.source", "global");
+  public static final CommonVars<String> NEBULA_USER_NAME =
+      CommonVars.apply("wds.linkis.nebula.username", "root");
 
-  public static final CommonVars<String> PRESTO_REQUEST_MEMORY =
-      CommonVars.apply("presto.session.query_max_total_memory", "8GB");
+  public static final CommonVars<String> NEBULA_PASSWORD =
+      CommonVars.apply("wds.linkis.nebula.password", "nebula");
 
-  public static final CommonVars<Boolean> PRESTO_SQL_HOOK_ENABLED =
-      CommonVars.apply("linkis.presto.sql.hook.enabled", true, "presto sql hook");
+  public static final CommonVars<Boolean> NEBULA_RECONNECT_ENABLED =
+      CommonVars.apply(
+          "linkis.nebula.reconnect.enabled",
+          false,
+          "whether to retry after the connection is disconnected");
 }
diff --git a/...src/main/java/org/apache/linkis/engineplugin/nebula/errorcode/NebulaErrorCodeSummary.java b/...src/main/java/org/apache/linkis/engineplugin/nebula/errorcode/NebulaErrorCodeSummary.java
@@ -21,9 +21,9 @@
 import org.apache.linkis.common.errorcode.LinkisErrorCode;
 
 public enum NebulaErrorCodeSummary implements LinkisErrorCode {
-  PRESTO_STATE_INVALID(
-      26001, "Presto status error,statement is not finished(Presto服务状态异常, 查询语句没有执行结束)"),
-  PRESTO_CLIENT_ERROR(26002, "Presto client error(Presto客户端异常)");
+  NEBULA_CLIENT_INITIALIZATION_FAILED(28001, "Nebula client initialization failed(Nebula客户端初始化失败)"),
+  NEBULA_EXECUTOR_ERROR(28002, "Nebula executor error(Nebula执行异常)"),
+  NEBULA_CLIENT_ERROR(28003, "Nebula client error(Nebula客户端异常)");
 
   private final int errorCode;