| /* |
| * Copyright 2019 WeBank |
| * |
| * Licensed 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 com.webank.wedatasphere.linkis.engine.execute |
| |
| import com.webank.wedatasphere.linkis.common.utils.{Logging, Utils} |
| import com.webank.wedatasphere.linkis.engine.conf.EngineConfiguration |
| import com.webank.wedatasphere.linkis.engine.extension.EnginePreExecuteHook |
| import com.webank.wedatasphere.linkis.resourcemanager.Resource |
| import com.webank.wedatasphere.linkis.scheduler.executer._ |
| import com.webank.wedatasphere.linkis.storage.resultset.ResultSetFactory |
| import org.apache.commons.lang.StringUtils |
| import org.apache.commons.lang.exception.ExceptionUtils |
| |
| import scala.collection.mutable.ArrayBuffer |
| |
| /** |
| * Created by enjoyyin on 2018/9/17. |
| */ |
| abstract class EngineExecutor(outputPrintLimit: Int, isSupportParallelism: Boolean) extends AbstractExecutor(0) with Logging { |
| |
| private val resultSetFactory = ResultSetFactory.getInstance |
| private var codeParser: Option[CodeParser] = None |
| private var resultSetListener: Option[ResultSetListener] = None |
| private var logListener: Option[JobLogListener] = None |
| private var jobProgressListener: Option[JobProgressListener] = None |
| private var engineInitialized = false |
| |
| private var executedNum = 0 |
| private var succeedNum = 0 |
| |
| |
| private val enginePreExecuteHooks:Array[EnginePreExecuteHook] = { |
| val hooks = new ArrayBuffer[EnginePreExecuteHook]() |
| EngineConfiguration.ENGINE_PRE_EXECUTE_HOOK_CLASSES.getValue.split(",") foreach { |
| hookStr => Utils.tryCatch{ |
| val clazz = Class.forName(hookStr.trim) |
| val obj = clazz.newInstance() |
| obj match { |
| case hook:EnginePreExecuteHook => hooks += hook |
| case _ => logger.warn(s"obj is not a engineHook obj is ${obj.getClass}") |
| } |
| }{ |
| case e:Exception => logger.error(s"failed to load class ${hookStr}") |
| } |
| } |
| hooks.toArray |
| } |
| |
| |
| def setCodeParser(codeParser: CodeParser) = this.codeParser = Some(codeParser) |
| def setResultSetListener(resultSetListener: ResultSetListener) = this.resultSetListener = Some(resultSetListener) |
| def getResultSetListener = resultSetListener |
| def setLogListener(logListener: JobLogListener) = this.logListener = Some(logListener) |
| def getLogListener = logListener |
| def setJobProgressListener(jobProgressListener: JobProgressListener) = this.jobProgressListener = Some(jobProgressListener) |
| def getJobProgressListener = jobProgressListener |
| |
| def init(): Unit = {} |
| |
| final def ready(): Unit = { |
| transition(ExecutorState.Idle) |
| if(!engineInitialized) { |
| engineInitialized = true |
| } |
| } |
| |
| def isEngineInitialized = engineInitialized |
| def getExecutedNum = executedNum |
| def getSucceedNum = succeedNum |
| def getFailedNum = executedNum - succeedNum |
| |
| def getName: String |
| |
| def getActualUsedResources: Resource |
| |
| def getDefaultResultSetType: String = resultSetFactory.getResultSetType(0) |
| protected def createEngineExecutorContext(executeRequest: ExecuteRequest): EngineExecutorContext = { |
| val engineExecutorContext = new EngineExecutorContext(this) |
| executeRequest match { |
| case job: JobExecuteRequest => engineExecutorContext.setJobId(job.jobId) |
| case _ => |
| } |
| executeRequest match { |
| case store: StorePathExecuteRequest => engineExecutorContext.setStorePath(store.storePath) |
| case _ => |
| } |
| engineExecutorContext |
| } |
| |
| protected def incompleteSplitter = "\n" |
| |
| protected def executeLine(engineExecutorContext: EngineExecutorContext, code: String): ExecuteResponse |
| protected def executeCompletely(engineExecutorContext: EngineExecutorContext, code: String, completedLine: String): ExecuteResponse |
| |
| override def execute(executeRequest: ExecuteRequest): ExecuteResponse = { |
| if (StringUtils.isEmpty(executeRequest.code)){ |
| return IncompleteExecuteResponse("Your code is incomplete, it may be that only comments are selected for execution(您的代码不完整，可能是仅仅选中了注释进行执行)") |
| } |
| executedNum += 1 |
| def ensureOp[A](f: => A): A = if(!this.engineInitialized) f |
| else if(isSupportParallelism) whenAvailable(f) else ensureIdle(f) |
| ensureOp { |
| val engineExecutorContext = createEngineExecutorContext(executeRequest) |
| var hookedCode = executeRequest.code; |
| Utils.tryCatch{ |
| enginePreExecuteHooks foreach { |
| hook => logger.info(s"${hook.hookName} begins to do a hook") |
| hookedCode = hook.callPreExecuteHook(engineExecutorContext, executeRequest, hookedCode) |
| logger.info(s"${hook.hookName} ends to do a hook") |
| } |
| }{ |
| case e:Throwable => logger.info("failed to do with hook", e) |
| } |
| info(s"hooked after code:$hookedCode") |
| var response: ExecuteResponse = null |
| val incomplete = new StringBuilder |
| val codes = Utils.tryCatch(codeParser.map(_.parse(hookedCode, engineExecutorContext)).getOrElse(Array(hookedCode))){ |
| e => info("Your code will be submitted in overall mode") |
| Array(hookedCode) |
| } |
| engineExecutorContext.setTotalParagraph(codes.length) |
| codes.indices.foreach { index => |
| if(engineExecutorContext.isKilled) return ErrorExecuteResponse("Job is killed by user!", null) |
| val code = codes(index) |
| engineExecutorContext.setCurrentParagraph(index + 1) |
| response = Utils.tryCatch(if(incomplete.nonEmpty) executeCompletely(engineExecutorContext, code, incomplete.toString) |
| else executeLine(engineExecutorContext, code)){ t => |
| ErrorExecuteResponse(ExceptionUtils.getRootCauseMessage(t), t) |
| } |
| incomplete ++= code |
| //engineExecutorContext.appendStdout(getName + ">> " + incomplete.toString().trim + " complete ") |
| response match { |
| case e: ErrorExecuteResponse => |
| error(s"execute code failed!", e.t) |
| return response |
| case SuccessExecuteResponse() => |
| engineExecutorContext.appendStdout("\n") |
| incomplete.setLength(0) |
| case e: OutputExecuteResponse => |
| incomplete.setLength(0) |
| val output = if(StringUtils.isNotEmpty(e.getOutput) && e.getOutput.length > outputPrintLimit) |
| e.getOutput.substring(0, outputPrintLimit) else e.getOutput |
| engineExecutorContext.appendStdout(output) |
| if(StringUtils.isNotBlank(e.getOutput)) engineExecutorContext.sendResultSet(e) |
| case _: IncompleteExecuteResponse => |
| incomplete ++= incompleteSplitter |
| } |
| } |
| Utils.tryCatch(engineExecutorContext.close()){ t => |
| response = ErrorExecuteResponse("send resultSet to entrance failed!", t) |
| } |
| response match { |
| case _: OutputExecuteResponse => |
| succeedNum += 1 |
| SuccessExecuteResponse() |
| case s: SuccessExecuteResponse => |
| succeedNum += 1 |
| s |
| case _ => response |
| } |
| } |
| } |
| |
| private[engine] def tryShutdown(): Unit = this.ensureAvailable(transition(ExecutorState.ShuttingDown)) |
| |
| private[engine] def tryDead(): Unit = this.whenState(ExecutorState.ShuttingDown, transition(ExecutorState.Dead)) |
| |
| override protected def callback(): Unit = {} |
| |
| // class EngineExecutorContext { |
| // private var jobId: Option[String] = None |
| // private val aliasNum = new AtomicInteger(0) |
| // protected var storePath: Option[String] = None |
| // |
| // def sendResultSet(resultSetWriter: ResultSetWriter[_ <: MetaData, _ <: Record]): Unit = { |
| // val fileName = new File(resultSetWriter.toFSPath.getPath).getName |
| // val index = if(fileName.indexOf(".") < 0) fileName.length else fileName.indexOf(".") |
| // val alias = if(fileName.startsWith("_")) fileName.substring(1, index) else fileName.substring(0, fileName.indexOf("_")) |
| // resultSetWriter.flush() |
| // Utils.tryFinally(sendResultSet(resultSetWriter.toString(), alias))(IOUtils.closeQuietly(resultSetWriter)) |
| // } |
| // |
| // def sendResultSet(output: String): Unit = sendResultSet(output, "_" + aliasNum.getAndIncrement()) |
| // |
| // private def sendResultSet(output: String, alias: String): Unit = |
| // if(resultSetFactory.isResultSetPath(output)) |
| // resultSetListener.foreach(l => jobId.foreach(l.onResultSetCreated(_, output, alias))) |
| // else if(resultSetFactory.isResultSet(output)) |
| // resultSetListener.foreach(l => jobId.foreach(l.onResultSetCreated(_, output, alias))) |
| // else throw new EngineErrorException(50050, "unknown resultSet: " + output) |
| // |
| // def setJobId(jobId: String) = this.jobId = Option(jobId) |
| // def getJobId = jobId |
| // def setStorePath(storePath: String) = this.storePath = Option(storePath) |
| // |
| // def sendResultSet(outputExecuteResponse: OutputExecuteResponse): Unit = outputExecuteResponse match { |
| // case AliasOutputExecuteResponse(alias, output) => sendResultSet(output, alias) |
| // case output: OutputExecuteResponse => sendResultSet(output.getOutput, "_" + aliasNum.getAndIncrement()) |
| // } |
| // |
| // protected def getDefaultStorePath: String = { |
| // val path = ENGINE_RESULT_SET_STORE_PATH.getValue |
| // (if(path.endsWith("/")) path else path + "/") + "user" + "/" + |
| // DateFormatUtils.format(System.currentTimeMillis(), "yyyyMMdd") + "/" + Sender.getThisModuleInstance.getApplicationName + |
| // "/" + Sender.getThisModuleInstance.getInstance + "/" + System.nanoTime |
| // } |
| // |
| // def createDefaultResultSetWriter(): ResultSetWriter[_ <: MetaData, _ <: Record] = createResultSetWriter(resultSetFactory.getResultSetByType(getDefaultResultSetType)) |
| // |
| // def createDefaultResultSetWriter(alias: String): ResultSetWriter[_ <: MetaData, _ <: Record] = |
| // createResultSetWriter(resultSetFactory.getResultSetByType(getDefaultResultSetType), alias) |
| // |
| // def createResultSetWriter(resultSetType: String): ResultSetWriter[_ <: MetaData, _ <: Record] = |
| // createResultSetWriter(resultSetFactory.getResultSetByType(resultSetType), null) |
| // |
| // def createResultSetWriter(resultSet: ResultSet[_ <: MetaData, _ <: Record]): ResultSetWriter[_ <: MetaData, _ <: Record] = |
| // createResultSetWriter(resultSet, null) |
| // |
| // def createResultSetWriter(resultSetType: String, alias: String): ResultSetWriter[_ <: MetaData, _ <: Record] = |
| // createResultSetWriter(resultSetFactory.getResultSetByType(resultSetType), alias) |
| // |
| // def createResultSetWriter(resultSet: ResultSet[_ <: MetaData, _ <: Record], alias: String): ResultSetWriter[_ <: MetaData, _ <: Record] = { |
| // val filePath = storePath.getOrElse(getDefaultStorePath) |
| // val fileName = if(StringUtils.isEmpty(alias)) "_" + aliasNum.getAndIncrement() else alias + "_" + aliasNum.getAndIncrement() |
| // val resultSetPath = resultSet.getResultSetPath(new FsPath(filePath), fileName) |
| // val resultSetWriter = ResultSetWriter.getResultSetWriter(resultSet, ENGINE_RESULT_SET_MAX_CACHE.getValue.toLong, resultSetPath) |
| // resultSetWriter |
| // } |
| // |
| // def appendStdout(log: String): Unit = if(!engineInitialized) |
| // info(log) else logListener.foreach(ll => jobId.foreach(ll.onLogUpdate(_, log))) |
| // } |
| } |
