Merge pull request #15 from jiaboy20/master

merge master to dev-fide
jiaboy20 · Nov 30, 2023 · 5bdd4a4 · 5bdd4a4
2 parents c3ad3f5 + 2347ae8
commit 5bdd4a4
Show file tree

Hide file tree

Showing 249 changed files with 14,101 additions and 1,444 deletions.
diff --git a/...ns/linkis-storage/src/main/java/org/apache/linkis/storage/exception/StorageErrorCode.java b/...ns/linkis-storage/src/main/java/org/apache/linkis/storage/exception/StorageErrorCode.java
@@ -23,7 +23,9 @@ public enum StorageErrorCode {
   FS_NOT_INIT(53001, "please init first"),
 
   INCONSISTENT_DATA(53001, "Inconsistent row data read,read %s,need rowLen %s"),
-  FS_OOM(53002, "OOM occurred while reading the file");
+  FS_OOM(53002, "OOM occurred while reading the file"),
+
+  FS_ERROR(53003, "Failed to operation fs");
 
   StorageErrorCode(int errorCode, String message) {
     this.code = errorCode;

diff --git a/...kis-storage/src/main/java/org/apache/linkis/storage/resultset/ResultSetReaderFactory.java b/...kis-storage/src/main/java/org/apache/linkis/storage/resultset/ResultSetReaderFactory.java
@@ -49,7 +49,7 @@ public static <K extends MetaData, V extends Record> ResultSetReader getResultSe
     return new StorageResultSetReader<>(resultSet, value);
   }
 
-  public static ResultSetReader getResultSetReader(String res) {
+  public static ResultSetReader getResultSetReader(String res) throws IOException {
     ResultSetFactory rsFactory = ResultSetFactory.getInstance();
     if (rsFactory.isResultSet(res)) {
       ResultSet<? extends MetaData, ? extends Record> resultSet = rsFactory.getResultSet(res);
@@ -58,21 +58,12 @@ public static ResultSetReader getResultSetReader(String res) {
       FsPath resPath = new FsPath(res);
       ResultSet<? extends MetaData, ? extends Record> resultSet =
           rsFactory.getResultSetByPath(resPath);
-      try {
-        FSFactory.getFs(resPath).init(null);
-      } catch (IOException e) {
-        logger.warn("ResultSetReaderFactory fs init failed", e);
-      }
-      ResultSetReader reader = null;
-      try {
-        reader =
-            ResultSetReaderFactory.getResultSetReader(
-                resultSet, FSFactory.getFs(resPath).read(resPath));
-      } catch (IOException e) {
-        logger.warn("ResultSetReaderFactory fs read failed", e);
-      }
+      Fs fs = FSFactory.getFs(resPath);
+      fs.init(null);
+      ResultSetReader reader =
+          ResultSetReaderFactory.getResultSetReader(resultSet, fs.read(resPath));
       if (reader instanceof StorageResultSetReader) {
-        ((StorageResultSetReader<?, ?>) reader).setFs(FSFactory.getFs(resPath));
+        ((StorageResultSetReader<?, ?>) reader).setFs(fs);
       }
       return (StorageResultSetReader<?, ?>) reader;
     }

diff --git a/...kis-storage/src/main/java/org/apache/linkis/storage/resultset/ResultSetWriterFactory.java b/...kis-storage/src/main/java/org/apache/linkis/storage/resultset/ResultSetWriterFactory.java
@@ -51,28 +51,25 @@ org.apache.linkis.common.io.resultset.ResultSetWriter<K, V> getResultSetWriter(
   public static Record[] getRecordByWriter(
       org.apache.linkis.common.io.resultset.ResultSetWriter<? extends MetaData, ? extends Record>
           writer,
-      long limit) {
+      long limit)
+      throws IOException {
     String res = writer.toString();
     return getRecordByRes(res, limit);
   }
 
-  public static Record[] getRecordByRes(String res, long limit) {
+  public static Record[] getRecordByRes(String res, long limit) throws IOException {
     ResultSetReader reader = ResultSetReaderFactory.getResultSetReader(res);
     int count = 0;
     List<Record> records = new ArrayList<>();
-    try {
-      reader.getMetaData();
-      while (reader.hasNext() && count < limit) {
-        records.add(reader.getRecord());
-        count++;
-      }
-    } catch (IOException e) {
-      logger.warn("ResultSetWriter getRecordByRes failed", e);
+    reader.getMetaData();
+    while (reader.hasNext() && count < limit) {
+      records.add(reader.getRecord());
+      count++;
     }
     return records.toArray(new Record[0]);
   }
 
-  public static Record getLastRecordByRes(String res) {
+  public static Record getLastRecordByRes(String res) throws IOException {
     ResultSetReader reader = ResultSetReaderFactory.getResultSetReader(res);
     Record record = null;
     try {

diff --git a/...kis-storage/src/main/java/org/apache/linkis/storage/resultset/StorageResultSetWriter.java b/...kis-storage/src/main/java/org/apache/linkis/storage/resultset/StorageResultSetWriter.java
@@ -24,6 +24,7 @@
 import org.apache.linkis.storage.*;
 import org.apache.linkis.storage.conf.*;
 import org.apache.linkis.storage.domain.*;
+import org.apache.linkis.storage.exception.StorageErrorException;
 import org.apache.linkis.storage.utils.*;
 
 import org.apache.commons.io.IOUtils;
@@ -37,6 +38,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.linkis.storage.exception.StorageErrorCode.FS_ERROR;
+
 public class StorageResultSetWriter<K extends MetaData, V extends Record>
     extends ResultSetWriter<K, V> {
   private static final Logger logger = LoggerFactory.getLogger(StorageResultSetWriter.class);
@@ -98,8 +101,9 @@ public void createNewFile() {
               fs.init(null);
               FileSystemUtils.createNewFile(storePath, proxyUser, true);
               outputStream = fs.write(storePath, true);
-            } catch (IOException e) {
-              logger.warn("StorageResultSetWriter createNewFile failed", e);
+            } catch (Exception e) {
+              throw new StorageErrorException(
+                  FS_ERROR.getCode(), "StorageResultSetWriter createNewFile failed", e);
             }
             logger.info("Succeed to create a new file:{}", storePath);
             fileCreated = true;

diff --git a/...ns/linkis-storage/src/main/java/org/apache/linkis/storage/source/ResultsetFileSource.java b/...ns/linkis-storage/src/main/java/org/apache/linkis/storage/source/ResultsetFileSource.java
@@ -47,7 +47,7 @@ record -> {
                             if (emptyValue.equals(Dolphin.LINKIS_NULL)) {
                               return "";
                             } else {
-                              return nullValue;
+                              return emptyValue;
                             }
                           } else if (r instanceof Double) {
                             return StorageUtils.doubleToString((Double) r);

diff --git a/...commons/linkis-storage/src/main/java/org/apache/linkis/storage/utils/FileSystemUtils.java b/...commons/linkis-storage/src/main/java/org/apache/linkis/storage/utils/FileSystemUtils.java
@@ -61,16 +61,12 @@ public static void createNewFile(FsPath filePath, boolean createParentWhenNotExi
     createNewFile(filePath, StorageUtils.getJvmUser(), createParentWhenNotExists);
   }
 
-  public static void createNewFile(
-      FsPath filePath, String user, boolean createParentWhenNotExists) {
+  public static void createNewFile(FsPath filePath, String user, boolean createParentWhenNotExists)
+      throws Exception {
     FileSystem fileSystem = (FileSystem) FSFactory.getFsByProxyUser(filePath, user);
     try {
       fileSystem.init(null);
       createNewFileWithFileSystem(fileSystem, filePath, user, createParentWhenNotExists);
-    } catch (IOException e) {
-      logger.warn("FileSystemUtils createNewFile failed", e);
-    } catch (Exception e) {
-      logger.warn("FileSystemUtils createNewFile failed", e);
     } finally {
       IOUtils.closeQuietly(fileSystem);
     }

diff --git a/...is-commons/linkis-storage/src/main/java/org/apache/linkis/storage/utils/StorageUtils.java b/...is-commons/linkis-storage/src/main/java/org/apache/linkis/storage/utils/StorageUtils.java
@@ -272,4 +272,8 @@ public static byte[] mergeByteArrays(byte[] arr1, byte[] arr2) {
     System.arraycopy(arr2, 0, mergedArray, arr1.length, arr2.length);
     return mergedArray;
   }
+
+  public static boolean isHDFSPath(FsPath fsPath) {
+    return HDFS.equalsIgnoreCase(fsPath.getFsType());
+  }
 }
diff --git a/...ance-common/src/main/java/org/apache/linkis/governance/common/constant/CodeConstants.java b/...ance-common/src/main/java/org/apache/linkis/governance/common/constant/CodeConstants.java
@@ -0,0 +1,23 @@
+/*
+ * 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 org.apache.linkis.governance.common.constant;
+
+public class CodeConstants {
+  // will auto append at end of scala code; make sure the last line is not a comment
+  public static String SCALA_CODE_AUTO_APPEND_CODE = "val linkisVar=123";
+}
diff --git a/...vernance-common/src/main/scala/org/apache/linkis/governance/common/paser/CodeParser.scala b/...vernance-common/src/main/scala/org/apache/linkis/governance/common/paser/CodeParser.scala
@@ -19,6 +19,7 @@ package org.apache.linkis.governance.common.paser
 
 import org.apache.linkis.common.utils.{CodeAndRunTypeUtils, Logging, Utils}
 import org.apache.linkis.governance.common.conf.GovernanceCommonConf
+import org.apache.linkis.governance.common.constant.CodeConstants
 import org.apache.linkis.governance.common.paser.CodeType.CodeType
 
 import org.apache.commons.lang3.StringUtils
@@ -116,7 +117,7 @@ class ScalaCodeParser extends SingleCodeParser with Logging {
     if (statementBuffer.nonEmpty) codeBuffer.append(statementBuffer.mkString("\n"))
     // Make sure the last line is not a comment
     codeBuffer.append("\n")
-    codeBuffer.append("val linkisVar=123")
+    codeBuffer.append(CodeConstants.SCALA_CODE_AUTO_APPEND_CODE)
     codeBuffer.toArray
   }
 

diff --git a/...nager-core/src/main/scala/org/apache/linkis/ecm/core/launch/ProcessEngineConnLaunch.scala b/...nager-core/src/main/scala/org/apache/linkis/ecm/core/launch/ProcessEngineConnLaunch.scala
@@ -176,8 +176,15 @@ trait ProcessEngineConnLaunch extends EngineConnLaunch with Logging {
       .findAvailPortByRange(GovernanceCommonConf.ENGINE_CONN_PORT_RANGE.getValue)
       .toString
 
-    var springConf = Map("server.port" -> engineConnPort, "spring.profiles.active" -> "engineconn")
-
+    var springConf =
+      Map[String, String]("server.port" -> engineConnPort, "spring.profiles.active" -> "engineconn")
+    val properties =
+      PortUtils.readFromProperties(Configuration.getLinkisHome + "/conf/version.properties")
+    if (StringUtils.isNotBlank(properties.getProperty("version"))) {
+      springConf += ("eureka.instance.metadata-map.linkis.app.version" -> properties.getProperty(
+        "version"
+      ))
+    }
     request.creationDesc.properties.asScala.filter(_._1.startsWith("spring.")).foreach {
       case (k, v) =>
         springConf = springConf + (k -> v)

diff --git a/...s-engineconn-manager-core/src/main/scala/org/apache/linkis/ecm/core/utils/PortUtils.scala b/...s-engineconn-manager-core/src/main/scala/org/apache/linkis/ecm/core/utils/PortUtils.scala
@@ -17,15 +17,16 @@
 
 package org.apache.linkis.ecm.core.utils
 
-import org.apache.linkis.common.utils.Utils
+import org.apache.linkis.common.utils.{Logging, Utils}
 
 import org.apache.commons.io.IOUtils
 import org.apache.commons.lang3.StringUtils
 
-import java.io.IOException
+import java.io.{BufferedReader, FileReader, IOException}
 import java.net.ServerSocket
+import java.util.Properties
 
-object PortUtils {
+object PortUtils extends Logging {
 
   /**
    * portRange: '-' is the separator
@@ -62,4 +63,23 @@ object PortUtils {
     Utils.tryFinally(socket.getLocalPort)(IOUtils.closeQuietly(socket))
   }
 
+  def readFromProperties(propertiesFile: String): Properties = {
+    val properties: Properties = new Properties
+    var reader: BufferedReader = null;
+    try {
+      reader = new BufferedReader(new FileReader(propertiesFile))
+      properties.load(reader)
+    } catch {
+      case e: Exception =>
+        logger.warn(s"loading vsersion faild with path $propertiesFile  error:$e")
+    } finally {
+      try if (reader != null) reader.close
+      catch {
+        case e: Exception =>
+          logger.warn(s"try to close buffered reader with error:${e.getMessage}")
+      }
+    }
+    properties
+  }
+
 }
diff --git a/...erver/src/main/java/org/apache/linkis/ecm/errorcode/EngineconnServerErrorCodeSummary.java b/...erver/src/main/java/org/apache/linkis/ecm/errorcode/EngineconnServerErrorCodeSummary.java
@@ -33,7 +33,15 @@ public enum EngineconnServerErrorCodeSummary implements LinkisErrorCode {
       11110,
       "the parameters of engineConnInstance and ticketId are both not exists.(engineConnInstance 和ticketId 的参数都不存在.)"),
   LOG_IS_NOT_EXISTS(11110, "Log directory {0} does not exists.(日志目录 {0} 不存在.)"),
-  FAILED_TO_DOWNLOAD(911115, "failed to downLoad(下载失败)");
+  FAILED_TO_DOWNLOAD(911115, "failed to downLoad(下载失败)"),
+  FILE_IS_OVERSIZE(911116, "Download file has exceeded 100MB(下载文件已超过100M)"),
+  PARAMETER_NOT_NULL(911117, "Parameter {0} cannot be empty (参数 {0} 不能为空)"),
+  LOGTYPE_ERROR(
+      911118,
+      "logType only supports stdout, stderr, gc, yarnApp(logType仅支持stdout,stderr,gc,yarnApp)"),
+  NOT_PERMISSION(
+      911119, "You {0} have no permission to download Log in ECM {1}(用户 {0} 无权限下载  ECM {1} 日志)"),
+  ;
 
   /** (errorCode)错误码 */
   private final int errorCode;