Merge pull request #54 from WeDataSphere/dev-0.2.3-log-collector

New log collector module

Author: jefftlin

pom.xml

@@ -45,14 +45,15 @@
     </modules>
 
     <properties>
-        <linkis.version>1.1.1</linkis.version>
+        <linkis.version>1.1.3</linkis.version>
+        <junit.version>4.12</junit.version>
         <dss.version>1.1.0</dss.version>
         <streamis.version>0.2.0</streamis.version>
         <scala.version>2.11.12</scala.version>
         <jdk.compile.version>1.8</jdk.compile.version>
         <maven.version>3.3.3</maven.version>
         <gson.version>2.8.5</gson.version>
-        <fasterxml.jackson.version>2.11.3</fasterxml.jackson.version>
+        <fasterxml.jackson.version>2.13.2</fasterxml.jackson.version>
         <math3.version>3.1.1</math3.version>
         <httpclient.version>4.5.4</httpclient.version>
         <httpmime.version>4.5.4</httpmime.version>

streamis-jobmanager/pom.xml

@@ -32,6 +32,7 @@
         <module>streamis-job-manager</module>
         <module>streamis-jobmanager-server</module>
         <module>streamis-projectmanager-server</module>
+        <module>streamis-job-log</module>
     </modules>
 
 

streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/java/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/url/LinkisURLStreamHandlerFactory.java

@@ -15,7 +15,6 @@
 
 package com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.url;
 
-import org.apache.commons.lang.StringUtils;
 
 import java.net.URLStreamHandler;
 import java.net.URLStreamHandlerFactory;

streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/core/FlinkLogIterator.scala

@@ -18,6 +18,7 @@ trait FlinkLogIterator extends Iterator[String] with Closeable {
   val engineConnLogOperator: EngineConnLogOperator
   def init(): Unit
   def getLogPath: String
+  def getLogDirSuffix: String
   def getLogs: util.ArrayList[String]
   def getEndLine: Long
 }
@@ -28,6 +29,7 @@ class SimpleFlinkJobLogIterator(override val requestPayload: LogRequestPayload,
   private var logs: util.ArrayList[String] = _
   private var index = 0
   private var logPath: String = _
+  private var logDirSuffix: String = _
   private var isClosed = true
   private var endLine = 0
 
@@ -69,4 +71,8 @@ class SimpleFlinkJobLogIterator(override val requestPayload: LogRequestPayload,
   override def getLogs: util.ArrayList[String] = logs
 
   override def getEndLine: Long = endLine
+
+  def setLogDirSuffix(logDirSuffix: String) : Unit = this.logDirSuffix = logDirSuffix
+
+  override def getLogDirSuffix: String = logDirSuffix
 }

streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/entity/LogRequestPayload.scala

@@ -14,6 +14,7 @@ class LogRequestPayload {
   private var onlyKeywords: String = _
   private var lastRows = 0
   private var logType: String = _
+  private var logHistory: Boolean = false
   def getPageSize: Int = pageSize
   def setPageSize(pageSize: Int): Unit = this.pageSize = pageSize
 
@@ -32,4 +33,8 @@ class LogRequestPayload {
   def getLogType: String = logType
 
   def setLogType(logType: String): Unit = this.logType = logType
+
+  def isLogHistory: Boolean = logHistory
+
+  def setLogHistory(logHistory: Boolean): Unit = this.logHistory = logHistory
 }

streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/job/FlinkJobClient.scala

@@ -23,13 +23,17 @@ import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.core.{FlinkLo
 import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.entity.LogRequestPayload
 import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.exception.{FlinkJobLaunchErrorException, FlinkJobStateFetchException, FlinkSavePointException}
 import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.manager.FlinkJobLaunchManager
-import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.operator.{FlinkTriggerSavepointOperator, FlinkYarnLogOperator}
+import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.operator.{FlinkClientLogOperator, FlinkTriggerSavepointOperator, FlinkYarnLogOperator}
 import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.state.{Checkpoint, Savepoint}
+import org.apache.commons.lang3.StringUtils
 import org.apache.linkis.common.utils.{Logging, Utils}
-import org.apache.linkis.computation.client.once.OnceJob
-import org.apache.linkis.computation.client.once.simple.SimpleOnceJob
+import org.apache.linkis.computation.client.once.action.ECResourceInfoAction
+import org.apache.linkis.computation.client.once.result.ECResourceInfoResult
+import org.apache.linkis.computation.client.once.{LinkisManagerClient, LinkisManagerClientImpl, OnceJob}
+import org.apache.linkis.computation.client.once.simple.{SimpleOnceJob, SimpleOnceJobBuilder}
 import org.apache.linkis.computation.client.operator.impl.EngineConnLogOperator
-
+import org.apache.linkis.httpclient.dws.DWSHttpClient
+import java.util
 import java.net.URI
 
 class FlinkJobClient(onceJob: OnceJob, var jobInfo: FlinkJobInfo, stateManager: JobStateManager)
@@ -39,9 +43,13 @@ class FlinkJobClient(onceJob: OnceJob, var jobInfo: FlinkJobInfo, stateManager:
    * Log operator
    */
   private var logOperatorMap = Map(
-    "client" -> EngineConnLogOperator.OPERATOR_NAME,
+    "client" -> FlinkClientLogOperator.OPERATOR_NAME,
     "yarn" -> FlinkYarnLogOperator.OPERATOR_NAME
   )
+  /**
+   * The linkis client in onceJob
+   */
+  private var linkisClient: DWSHttpClient = _
 
   override def getJobInfo: FlinkJobInfo = {
     getJobInfo(false)
@@ -99,13 +107,38 @@ class FlinkJobClient(onceJob: OnceJob, var jobInfo: FlinkJobInfo, stateManager:
       case Some(operator) =>
         onceJob.getOperator(operator) match {
           case engineConnLogOperator: EngineConnLogOperator =>
+            val logIterator = new SimpleFlinkJobLogIterator(requestPayload, engineConnLogOperator)
+            engineConnLogOperator match {
+              case clientLogOperator: FlinkClientLogOperator =>
+                var logDirSuffix = this.jobInfo.getLogDirSuffix
+                if (StringUtils.isBlank(logDirSuffix) && requestPayload.isLogHistory){
+                  // If want to fetch the history log, must get the log directory suffix first
+                  getLinkisClient match {
+                    case client: DWSHttpClient =>
+                      Option(Utils.tryCatch{
+                        client.execute(ECResourceInfoAction.newBuilder().setUser(jobInfo.getUser)
+                          .setTicketid(clientLogOperator.getTicketId).build()).asInstanceOf[ECResourceInfoResult]
+                      }{
+                        case e: Exception =>
+                          warn("Fail to query the engine conn resource info from linkis",  e)
+                          null
+                      }) match {
+                        case Some(result) => logDirSuffix = Utils.tryAndWarn{result.getData.getOrDefault("ecResourceInfoRecord", new util.HashMap[String, Any]).asInstanceOf[util.Map[String, Any]]
+                          .getOrDefault("logDirSuffix", "").asInstanceOf[String]}
+                        case _ =>
+                      }
+                  }
+                }
+                clientLogOperator.setLogDirSuffix(logDirSuffix)
+                logIterator.setLogDirSuffix(logDirSuffix)
+              case _ =>
+            }
             engineConnLogOperator match {
               case yarnLogOperator: FlinkYarnLogOperator => yarnLogOperator.setApplicationId(jobInfo.getApplicationId)
               case _ =>
             }
             engineConnLogOperator.setECMServiceInstance(jobInfo.getECMInstance)
             engineConnLogOperator.setEngineConnType(FlinkJobLaunchManager.FLINK_ENGINE_CONN_TYPE)
-            val logIterator = new SimpleFlinkJobLogIterator(requestPayload, engineConnLogOperator)
             logIterator.init()
             jobInfo match {
               case jobInfo: FlinkJobInfo => jobInfo.setLogPath(logIterator.getLogPath)
@@ -161,5 +194,27 @@ class FlinkJobClient(onceJob: OnceJob, var jobInfo: FlinkJobInfo, stateManager:
     triggerSavepoint(savepointURI.toString, JobLauncherConfiguration.FLINK_TRIGGER_SAVEPOINT_MODE.getValue)
   }
 
+  /**
+   * Get linkis client
+   * @return
+   */
+  def getLinkisClient: DWSHttpClient = {
+    Utils.tryAndWarn{
+      if (null == this.linkisClient){
+        this.synchronized{
+            if (null == this.linkisClient){
+              this.linkisClient = SimpleOnceJobBuilder.getLinkisManagerClient match {
+                case client: LinkisManagerClient =>
+                  val dwsClientField = classOf[LinkisManagerClientImpl].getDeclaredField("dwsHttpClient")
+                  dwsClientField.setAccessible(true)
+                  dwsClientField.get(client).asInstanceOf[DWSHttpClient]
+                case _ => null
+              }
 
+            }
+        }
+      }
+      this.linkisClient
+    }
+  }
 }

streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/job/FlinkJobInfo.scala

@@ -34,41 +34,53 @@ class FlinkJobInfo extends YarnJobInfo {
   private var applicationUrl: String = _
   private var status: String = _
   private var logPath: String = _
+  private var logDirSuffix: String = _
   private var resources: java.util.Map[String, Object] = _
   private var completedMsg: String = _
   private var jobStates: Array[JobStateInfo] = _
+
   override def getApplicationId: String = applicationId
+
   def setApplicationId(applicationId: String): Unit = this.applicationId = applicationId
 
   override def getApplicationUrl: String = applicationUrl
+
   def setApplicationUrl(applicationUrl: String): Unit = this.applicationUrl = applicationUrl
 
   override def getId: String = id
-  def setId(id: String): Unit = this.id = id
 
+  def setId(id: String): Unit = this.id = id
 
   override def getECMInstance: ServiceInstance = ecmInstance
+
   def setECMInstance(ecmInstance: ServiceInstance): Unit = this.ecmInstance = ecmInstance
 
   override def getUser: String = user
+
   def setUser(user: String): Unit = this.user = user
 
   override def getStatus: String = status
+
   override def setStatus(status: String): Unit = this.status = status
 
   override def getLogPath: String = logPath
+
   def setLogPath(logPath: String): Unit = this.logPath = logPath
 
   override def getResources: util.Map[String, Object] = resources
+
   def setResources(resources: java.util.Map[String, Object]): Unit = this.resources = resources
 
   def getSavepoint: String = savepoint
+
   def setSavepoint(savepoint: String): Unit = this.savepoint = savepoint
 
   def getCheckpoint: String = checkpoint
+
   def setCheckpoint(checkpoint: String): Unit = this.checkpoint = checkpoint
 
   override def getCompletedMsg: String = completedMsg
+
   def setCompletedMsg(completedMsg: String): Unit = this.completedMsg = completedMsg
 
   override def toString: String = s"FlinkJobInfo(id: $id, status: $status, applicationId: $applicationId, applicationUrl: $applicationUrl, logPath: $logPath)"
@@ -85,6 +97,7 @@ class FlinkJobInfo extends YarnJobInfo {
   def setJobStates(jobStates: Array[JobStateInfo]): Unit = {
     this.jobStates = jobStates
   }
+
   /**
    * Job name
    *
@@ -95,11 +108,16 @@ class FlinkJobInfo extends YarnJobInfo {
   def setName(name: String): Unit = {
     this.name = name
   }
-}
 
-object FlinkJobInfo{
-  def main(args: Array[String]): Unit = {
-    val jobInfo = "{\"jobStates:\":{\"location\":\"xx\"}"
-    DWSHttpClient.jacksonJson.readValue(jobInfo, classOf[FlinkJobInfo])
+  /**
+   * Job log directory suffix
+   *
+   * @return
+   */
+  override def getLogDirSuffix: String = this.logDirSuffix
+
+  override def setLogDirSuffix(logDirSuffix: String): Unit = {
+    this.logDirSuffix = logDirSuffix
   }
 }
+

streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/job/LinkisJobInfo.scala

@@ -11,5 +11,11 @@ trait LinkisJobInfo extends JobInfo {
    */
   def getECMInstance: ServiceInstance
 
+  /**
+   * Job log directory suffix
+   * @return
+   */
+  def getLogDirSuffix: String
 
+  def setLogDirSuffix(logDirSuffix: String): Unit
 }

streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/job/manager/SimpleFlinkJobLaunchManager.scala

@@ -15,16 +15,16 @@
 
 package com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.manager
 
-import com.webank.wedatasphere.streamis.jobmanager.launcher.job.state.{JobState, JobStateInfo}
+import com.webank.wedatasphere.streamis.jobmanager.launcher.job.state.JobState
 import com.webank.wedatasphere.streamis.jobmanager.launcher.job.{JobClient, LaunchJob}
 import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.conf.JobLauncherConfiguration
-import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.{FlinkJobClient, FlinkJobInfo, LinkisJobInfo}
 import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.manager.SimpleFlinkJobLaunchManager.INSTANCE_NAME
-import org.apache.commons.lang.StringEscapeUtils
+import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.{FlinkJobClient, FlinkJobInfo, LinkisJobInfo}
+import org.apache.commons.lang3.StringEscapeUtils
 import org.apache.linkis.common.utils.{RetryHandler, Utils}
 import org.apache.linkis.computation.client.once.simple.{SimpleOnceJob, SubmittableSimpleOnceJob}
 import org.apache.linkis.computation.client.once.{OnceJob, SubmittableOnceJob}
-import org.apache.linkis.computation.client.operator.impl.{EngineConnApplicationInfoOperator, EngineConnLogOperator}
+import org.apache.linkis.computation.client.operator.impl.EngineConnApplicationInfoOperator
 import org.apache.linkis.httpclient.dws.DWSHttpClient
 import org.apache.linkis.ujes.client.exception.UJESJobException
 

streamis-jobmanager/streamis-job-launcher/streamis-job-launcher-linkis/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/launcher/linkis/job/operator/FlinkClientLogOperator.scala

@@ -0,0 +1,34 @@
+package com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.operator
+
+import org.apache.commons.lang3.StringUtils
+import org.apache.linkis.computation.client.once.action.EngineConnOperateAction
+import org.apache.linkis.computation.client.operator.impl.{EngineConnLogOperator, EngineConnLogs}
+
+/**
+ * Append "logDirSuffix" parameter
+ */
+class FlinkClientLogOperator extends EngineConnLogOperator{
+
+  private var logDirSuffix: String = _
+
+  def setLogDirSuffix(logDirSuffix: String): Unit = {
+    this.logDirSuffix = logDirSuffix
+  }
+
+  protected override def addParameters(builder: EngineConnOperateAction.Builder): Unit = {
+    builder.operatorName(EngineConnLogOperator.OPERATOR_NAME)
+    if (StringUtils.isNotBlank(this.logDirSuffix)) {
+      builder.addParameter("logDirSuffix", logDirSuffix)
+    }
+    super.addParameters(builder)
+  }
+
+
+  override def getTicketId: String = super.getTicketId
+
+  override def getName: String = FlinkClientLogOperator.OPERATOR_NAME
+}
+
+object FlinkClientLogOperator {
+  val OPERATOR_NAME = "engineConnLog_flink"
+}