Reduce mixed language module linkismange #4463

Reduce mixed language module linkismange #4463

linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/exception/ErrorException.java

@@ -17,7 +17,7 @@
 
 package org.apache.linkis.common.exception;
 
-public class ErrorException extends LinkisException {
+public class ErrorException extends LinkisRuntimeException {
   private ExceptionLevel level = ExceptionLevel.ERROR;
 
   public ErrorException(int errCode, String desc) {

linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/exception/FatalException.java

@@ -17,7 +17,7 @@
 
 package org.apache.linkis.common.exception;
 
-public class FatalException extends LinkisException {
+public class FatalException extends LinkisRuntimeException {
   private ExceptionLevel level = ExceptionLevel.FATAL;
 
   public FatalException(int errCode, String desc) {

linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/exception/LinkisRetryException.java

@@ -17,7 +17,7 @@
 
 package org.apache.linkis.common.exception;
 
-public class LinkisRetryException extends LinkisException {
+public class LinkisRetryException extends LinkisRuntimeException {
   LinkisRetryException(int errCode, String desc, String ip, int port, String serviceKind) {
     super(errCode, desc, ip, port, serviceKind);
   }
@@ -27,7 +27,7 @@ public LinkisRetryException(int errCode, String desc) {
   }
 
   @Override
-  ExceptionLevel getLevel() {
+  public ExceptionLevel getLevel() {
     return ExceptionLevel.RETRY;
   }
 }

linkis-computation-governance/linkis-computation-governance-common/src/main/java/org/apache/linkis/governance/common/utils/ECPathUtils.java

@@ -0,0 +1,53 @@
+/*
+ * 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.utils;
+
+import org.apache.linkis.manager.label.entity.engine.EngineTypeLabel;
+import org.apache.linkis.manager.label.entity.engine.UserCreatorLabel;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.time.DateFormatUtils;
+
+import java.io.File;
+import java.nio.file.Paths;
+
+public class ECPathUtils {
+
+  public static String getECWOrkDirPathSuffix(String user, String ticketId, String engineType) {
+    String engineTypeRes = "";
+    if (StringUtils.isNotBlank(engineType)) {
+      engineTypeRes = engineType;
+    }
+    File file =
+        Paths.get(
+                user, DateFormatUtils.format(System.currentTimeMillis(), "yyyyMMdd"), engineTypeRes)
+            .toFile();
+    return file.getPath() + File.separator + ticketId;
+  }
+
+  public static String getECLogDirSuffix(
+      EngineTypeLabel engineTypeLabel, UserCreatorLabel userCreatorLabel, String ticketId) {
+    if (null == engineTypeLabel || null == userCreatorLabel) {
+      return "";
+    }
+    String ecwOrkDirPathSuffix =
+        ECPathUtils.getECWOrkDirPathSuffix(
+            userCreatorLabel.getUser(), ticketId, engineTypeLabel.getEngineType());
+    return ecwOrkDirPathSuffix + File.separator + "logs";
+  }
+}

linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/exception/GovernanceErrorException.scala

@@ -17,14 +17,15 @@
 
 package org.apache.linkis.governance.common.exception
 
-import org.apache.linkis.common.exception.ErrorException
+import org.apache.linkis.common.exception.{ErrorException, ExceptionLevel, LinkisRuntimeException}
 
 class GovernanceErrorException(errorCode: Int, errorMsg: String)
-    extends ErrorException(errorCode, errorMsg) {
+    extends LinkisRuntimeException(errorCode, errorMsg) {
 
   def this(errorCode: Int, errorMsg: String, cause: Throwable) = {
     this(errorCode, errorMsg)
     initCause(cause)
   }
 
+  override def getLevel: ExceptionLevel = ExceptionLevel.ERROR
 }

linkis-computation-governance/linkis-engineconn-manager/linkis-engineconn-manager-server/src/main/java/org/apache/linkis/ecm/server/operator/EngineConnLogOperator.java

@@ -0,0 +1,219 @@
+/*
+ * 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.ecm.server.operator;
+
+import org.apache.linkis.common.conf.CommonVars;
+import org.apache.linkis.common.utils.Utils;
+import org.apache.linkis.ecm.server.conf.ECMConfiguration;
+import org.apache.linkis.ecm.server.exception.ECMErrorException;
+import org.apache.linkis.manager.common.operator.Operator;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.io.input.ReversedLinesFileReader;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang3.tuple.Triple;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.text.MessageFormat;
+import java.util.*;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.linkis.ecm.errorcode.EngineconnServerErrorCodeSummary.*;
+
+public class EngineConnLogOperator implements Operator {
+  private static final Logger logger = LoggerFactory.getLogger(EngineConnLogOperator.class);
+
+  public static final String OPERATOR_NAME = "engineConnLog";
+  public static final CommonVars<String> LOG_FILE_NAME =
+      CommonVars.apply("linkis.engineconn.log.filename", "stdout");
+  public static final CommonVars<Integer> MAX_LOG_FETCH_SIZE =
+      CommonVars.apply("linkis.engineconn.log.fetch.lines.max", 5000);
+  public static final CommonVars<Integer> MAX_LOG_TAIL_START_SIZE =
+      CommonVars.apply("linkis.engineconn.log.tail.start.size");
+  public static final CommonVars<String> MULTILINE_PATTERN =
+      CommonVars.apply(
+          "linkis.engineconn.log.multiline.pattern",
+          "^\\d{4}-\\d{2}-\\d{2}\\s+\\d{2}:\\d{2}:\\d{2}\\.\\d{3}");
+  public static final CommonVars<Integer> MULTILINE_MAX =
+      CommonVars.apply("linkis.engineconn.log.multiline.max", 500);
+
+  @Override
+  public String[] getNames() {
+    return new String[] {OPERATOR_NAME};
+  }
+
+  @Override
+  public Map<String, Object> apply(Map<String, Object> parameters) {
+    File logPath = getLogPath(parameters);
+    int lastRows = getAs(parameters, "lastRows", 0);
+    int pageSize = getAs(parameters, "pageSize", 100);
+    int fromLine = getAs(parameters, "fromLine", 1);
+    boolean enableTail = getAs(parameters, "enableTail", false);
+    if (lastRows > EngineConnLogOperator.MAX_LOG_FETCH_SIZE.getValue()) {
+      throw new ECMErrorException(
+          CANNOT_FETCH_MORE_THAN.getErrorCode(),
+          MessageFormat.format(
+              CANNOT_FETCH_MORE_THAN.getErrorDesc(),
+              EngineConnLogOperator.MAX_LOG_FETCH_SIZE.getValue().toString()));
+    } else if (lastRows > 0) {
+      String logs = Utils.exec(new String[] {"tail", "-n", lastRows + "", logPath.getPath()}, 5000);
+      Map<String, Object> stringObjectHashMap = new HashMap<>();
+      stringObjectHashMap.put("logs", logs.split("\n"));
+      stringObjectHashMap.put("rows", logs.length());
+      return stringObjectHashMap;
+    }
+
+    String ignoreKeywords = getAs(parameters, "ignoreKeywords", "");
+    String[] ignoreKeywordList =
+        StringUtils.isNotEmpty(ignoreKeywords) ? ignoreKeywords.split(",") : new String[0];
+
+    String onlyKeywords = getAs(parameters, "onlyKeywords", "");
+    String[] onlyKeywordList =
+        StringUtils.isNotEmpty(onlyKeywords) ? onlyKeywords.split(",") : new String[0];
+
+    RandomAccessFile randomReader = null;
+    ReversedLinesFileReader reversedReader = null;
+    try {
+      if (enableTail) {
+        logger.info("enable log operator from tail to read");
+        reversedReader = new ReversedLinesFileReader(logPath, Charset.defaultCharset());
+      } else {
+        randomReader = new RandomAccessFile(logPath, "r");
+      }
+
+      ArrayList<String> logs = new ArrayList<>(pageSize);
+      int readLine = 0, skippedLine = 0, lineNum = 0;
+      boolean rowIgnore = false;
+      int ignoreLine = 0;
+      Pattern linePattern = Pattern.compile(EngineConnLogOperator.MULTILINE_PATTERN.getValue());
+
+      int maxMultiline = MULTILINE_MAX.getValue();
+      String line = randomAndReversedReadLine(randomReader, reversedReader);
+
+      while (readLine < pageSize && line != null) {
+        lineNum += 1;
+        if (skippedLine < fromLine - 1) {
+          skippedLine += 1;
+        } else {
+          if (rowIgnore) {
+            Matcher matcher = linePattern.matcher(line);
+            if (matcher.matches()) {
+              ignoreLine = 0;
+              rowIgnore = !includeLine(line, onlyKeywordList, ignoreKeywordList);
+            } else {
+              ignoreLine += 1;
+              if (ignoreLine >= maxMultiline) {
+                rowIgnore = false;
+              }
+            }
+            if (!matcher.matches()) {
+              rowIgnore = !includeLine(line, onlyKeywordList, ignoreKeywordList);
+            }
+          } else {
+            rowIgnore = !includeLine(line, onlyKeywordList, ignoreKeywordList);
+          }
+          if (!rowIgnore) {
+            logs.add(line);
+            readLine += 1;
+          }
+        }
+        line = randomAndReversedReadLine(randomReader, reversedReader);
+      }
+
+      if (enableTail) {
+        Collections.reverse(logs);
+      }
+
+      Map<String, Object> resultMap = new HashMap<>();
+      resultMap.put("logPath", logPath.getPath());
+      resultMap.put("logs", logs);
+      resultMap.put("endLine", lineNum);
+      resultMap.put("rows", readLine);
+      return resultMap;
+    } catch (IOException e) {
+      logger.info("EngineConnLogOperator apply failed", e);
+      throw new ECMErrorException(
+          LOG_IS_NOT_EXISTS.getErrorCode(), LOG_IS_NOT_EXISTS.getErrorDesc());
+    } finally {
+      IOUtils.closeQuietly(randomReader);
+      IOUtils.closeQuietly(reversedReader);
+    }
+  }
+
+  private String randomAndReversedReadLine(
+      RandomAccessFile randomReader, ReversedLinesFileReader reversedReader) throws IOException {
+    if (randomReader != null) {
+      String line = randomReader.readLine();
+      if (line != null) {
+        return new String(line.getBytes(StandardCharsets.ISO_8859_1), Charset.defaultCharset());
+      } else {
+        return null;
+      }
+    } else {
+      return reversedReader.readLine();
+    }
+  }
+
+  protected File getLogPath(Map<String, Object> parameters) {
+    String logType = getAs(parameters, "logType", EngineConnLogOperator.LOG_FILE_NAME.getValue());
+
+    Triple<String, String, String> engineConnInfo = getEngineConnInfo(parameters);
+    String engineConnLogDir = engineConnInfo.getLeft();
+    String engineConnInstance = engineConnInfo.getMiddle();
+    String ticketId = engineConnInfo.getRight();
+
+    File logPath = new File(engineConnLogDir, logType);
+    if (!logPath.exists() || !logPath.isFile()) {
+      throw new ECMErrorException(
+          LOGFILE_IS_NOT_EXISTS.getErrorCode(),
+          MessageFormat.format(LOGFILE_IS_NOT_EXISTS.getErrorDesc(), logPath.toString()));
+    }
+    logger.info(
+        String.format(
+            "Try to fetch EngineConn(id: %s, instance: %s) logs from %s.",
+            ticketId, engineConnInstance, logPath.getPath()));
+    return logPath;
+  }
+
+  protected Triple<String, String, String> getEngineConnInfo(Map<String, Object> parameters) {
+    String logDIrSuffix = getAs(parameters, "logDirSuffix", "");
+    String engineConnLogDir =
+        ECMConfiguration.ENGINECONN_ROOT_DIR() + File.separator + logDIrSuffix;
+    String ticketId = getAs(parameters, "ticketId", "");
+    String engineConnInstance = "";
+    return Triple.of(engineConnLogDir, engineConnInstance, ticketId);
+  }
+
+  private boolean includeLine(String line, String[] onlyKeywordList, String[] ignoreKeywordList) {
+    boolean accept =
+        ignoreKeywordList.length == 0 || !Arrays.stream(ignoreKeywordList).anyMatch(line::contains);
+    if (accept) {
+      accept =
+          onlyKeywordList.length == 0 || Arrays.stream(onlyKeywordList).anyMatch(line::contains);
+    }
+    return accept;
+  }
+}