You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@linkis.apache.org by pe...@apache.org on 2022/06/07 12:44:46 UTC
[incubator-linkis] branch dev-1.1.3 updated (adfc90243 -> 16f645bb8)
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a change to branch dev-1.1.3
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
from adfc90243 linkis-entranc - fix JobQueueLabel and jobRunningLabel (#2237)
new 5c1ef13ce optimize TaskYarnResourceEvent and TaskProgressEvent as one Event, defined as TaskRunningInfoEvent: 1. Record resource information 2. Record progress information 3. Record ec information
new f49cd0759 linkis-computation-orchestrator - fix
new 27f91caea 1. fix DefaultProgressOperation
new f0780ad11 1. linkis-entrance - fix method progressWithResource
new b42554fa7 1. linkis-module - add default logger in mybatis-config.xml 2. linkis-entrance - fix resource NPE
new 504b2f71e 1. linkis-jobhistory - add engineinstance
new 185f027f1 format code
new 229cb020a rename TaskConstants to TaskInfoConstants
new 16f645bb8 Fix NPE
The 9 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../src/main/resources/mybatis-config.xml | 2 +
.../linkis/protocol/constants/TaskConstant.java | 3 +
.../common/constant/job/TaskInfoConstants.scala | 14 ++--
.../service/TaskExecutionServiceImpl.scala | 25 +++++--
.../entrance/restful/EntranceRestfulApi.java | 77 ++++++++++++----------
.../entrance/execute/DefaultEntranceExecutor.scala | 59 ++++++++++-------
.../linkis/entrance/execute/EntranceExecutor.scala | 8 +--
.../EntranceOrchestrationFactory.scala | 1 -
.../protocol/resource/ResourceProtocol.scala | 4 ++
.../progress/AbstractProgressOperation.scala | 18 ++---
.../progress/DefaultProgressOperation.scala | 52 +++++++++++----
.../operation/progress/ProgressInfoEvent.scala | 6 +-
.../operation/progress/ProgressProcessor.scala | 13 ++--
.../service/ComputationTaskExecutionReceiver.scala | 19 ++----
.../orchestrator/listener/task/TaskInfoEvent.scala | 3 +-
.../listener/task/TaskProgressListener.scala | 2 +-
.../plans/logical/LogicalContext.scala | 4 +-
.../plans/logical/LogicalContextImpl.scala | 4 +-
.../plans/physical/PhysicalContext.scala | 6 +-
.../plans/physical/PhysicalContextImpl.scala | 9 +--
.../ecm/service/TaskExecutionReceiver.scala | 6 +-
.../jobhistory/conversions/TaskConversions.scala | 16 +++--
22 files changed, 203 insertions(+), 148 deletions(-)
copy linkis-public-enhancements/linkis-datasource/linkis-metadata/src/main/java/org/apache/linkis/metadata/hive/config/DataSource.java => linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/constant/job/TaskInfoConstants.scala (81%)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[incubator-linkis] 05/09: 1. linkis-module - add default logger in mybatis-config.xml 2. linkis-entrance - fix resource NPE
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch dev-1.1.3
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
commit b42554fa70cf8d6da8f19b11a3235fe120363af1
Author: alexkun <xu...@qq.com>
AuthorDate: Tue May 31 23:14:20 2022 +0800
1. linkis-module - add default logger in mybatis-config.xml
2. linkis-entrance - fix resource NPE
---
.../src/main/resources/mybatis-config.xml | 2 +
.../entrance/restful/EntranceRestfulApi.java | 66 +++++++++++-----------
2 files changed, 36 insertions(+), 32 deletions(-)
diff --git a/linkis-commons/linkis-module/src/main/resources/mybatis-config.xml b/linkis-commons/linkis-module/src/main/resources/mybatis-config.xml
index a55c1cf20..71a2f0e37 100644
--- a/linkis-commons/linkis-module/src/main/resources/mybatis-config.xml
+++ b/linkis-commons/linkis-module/src/main/resources/mybatis-config.xml
@@ -49,6 +49,8 @@
<!-- 默认为OTHER,为了解决oracle插入null报错的问题要设置为NULL -->
<setting name="jdbcTypeForNull" value="NULL" />
<setting name="lazyLoadTriggerMethods" value="equals,clone,hashCode,toString" />
+ <!-- 开启日志 -->
+ <setting name="logImpl" value="LOG4J2" />
</settings>
</configuration>
diff --git a/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java b/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java
index 11d8958f3..62f425b16 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java
+++ b/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java
@@ -300,39 +300,41 @@ public class EntranceRestfulApi implements EntranceRestfulRemote {
resoureList.add(
new YarnResourceWithStatusVo(applicationId, resource));
});
- }
- metricsVo.put(TaskConstant.ENTRANCEJOB_YARNRESOURCE, resoureList);
- Optional<Integer> cores =
- resourceMap.values().stream()
- .map(resource -> resource.queueCores())
- .reduce((x, y) -> x + y);
- Optional<Long> memory =
- resourceMap.values().stream()
- .map(resource -> resource.queueMemory())
- .reduce((x, y) -> x + y);
- float corePercent = 0.0f;
- float memoryPercent = 0.0f;
- if (cores.isPresent()) {
- corePercent =
- cores.get().floatValue()
- / EntranceConfiguration.YARN_QUEUE_CORES_MAX().getValue();
- memoryPercent =
- memory.get().floatValue()
- / (EntranceConfiguration.YARN_QUEUE_MEMORY_MAX()
- .getValue()
- .longValue()
- * 1024
- * 1024
- * 1024);
- }
- String coreRGB = RGBUtils.getRGB(corePercent);
- String memoryRGB = RGBUtils.getRGB(memoryPercent);
- metricsVo.put(TaskConstant.ENTRANCEJOB_CORE_PERCENT, corePercent);
- metricsVo.put(TaskConstant.ENTRANCEJOB_MEMORY_PERCENT, memoryPercent);
- metricsVo.put(TaskConstant.ENTRANCEJOB_CORE_RGB, coreRGB);
- metricsVo.put(TaskConstant.ENTRANCEJOB_MEMORY_RGB, memoryRGB);
+ metricsVo.put(TaskConstant.ENTRANCEJOB_YARNRESOURCE, resoureList);
+ Optional<Integer> cores =
+ resourceMap.values().stream()
+ .map(resource -> resource.queueCores())
+ .reduce((x, y) -> x + y);
+ Optional<Long> memory =
+ resourceMap.values().stream()
+ .map(resource -> resource.queueMemory())
+ .reduce((x, y) -> x + y);
+ float corePercent = 0.0f;
+ float memoryPercent = 0.0f;
+ if (cores.isPresent()) {
+ corePercent =
+ cores.get().floatValue()
+ / EntranceConfiguration.YARN_QUEUE_CORES_MAX().getValue();
+ memoryPercent =
+ memory.get().floatValue()
+ / (EntranceConfiguration.YARN_QUEUE_MEMORY_MAX()
+ .getValue()
+ .longValue()
+ * 1024
+ * 1024
+ * 1024);
+ }
+ String coreRGB = RGBUtils.getRGB(corePercent);
+ String memoryRGB = RGBUtils.getRGB(memoryPercent);
+ metricsVo.put(TaskConstant.ENTRANCEJOB_CORE_PERCENT, corePercent);
+ metricsVo.put(TaskConstant.ENTRANCEJOB_MEMORY_PERCENT, memoryPercent);
+ metricsVo.put(TaskConstant.ENTRANCEJOB_CORE_RGB, coreRGB);
+ metricsVo.put(TaskConstant.ENTRANCEJOB_MEMORY_RGB, memoryRGB);
- message.data(TaskConstant.ENTRANCEJOB_YARN_METRICS, metricsVo);
+ message.data(TaskConstant.ENTRANCEJOB_YARN_METRICS, metricsVo);
+ } else {
+ message.data(TaskConstant.ENTRANCEJOB_YARNRESOURCE, null);
+ }
} else {
message.data(TaskConstant.ENTRANCEJOB_YARNRESOURCE, null);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[incubator-linkis] 06/09: 1. linkis-jobhistory - add engineinstance
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch dev-1.1.3
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
commit 504b2f71e87e7c3ca30592d84429d11012ace160
Author: alexkun <xu...@qq.com>
AuthorDate: Tue May 31 23:30:13 2022 +0800
1. linkis-jobhistory - add engineinstance
---
.../linkis/jobhistory/conversions/TaskConversions.scala | 16 ++++++++++++----
1 file changed, 12 insertions(+), 4 deletions(-)
diff --git a/linkis-public-enhancements/linkis-publicservice/linkis-jobhistory/src/main/scala/org/apache/linkis/jobhistory/conversions/TaskConversions.scala b/linkis-public-enhancements/linkis-publicservice/linkis-jobhistory/src/main/scala/org/apache/linkis/jobhistory/conversions/TaskConversions.scala
index 4f3f3e55e..92546d3b6 100644
--- a/linkis-public-enhancements/linkis-publicservice/linkis-jobhistory/src/main/scala/org/apache/linkis/jobhistory/conversions/TaskConversions.scala
+++ b/linkis-public-enhancements/linkis-publicservice/linkis-jobhistory/src/main/scala/org/apache/linkis/jobhistory/conversions/TaskConversions.scala
@@ -230,7 +230,7 @@ object TaskConversions extends Logging {
taskVO.setInstance(job.getInstances)
taskVO.setExecId(job.getJobReqId)
taskVO.setUmUser(job.getSubmitUser)
- taskVO.setEngineInstance(null)
+
taskVO.setProgress(job.getProgress)
taskVO.setLogPath(job.getLogPath)
taskVO.setStatus(job.getStatus)
@@ -266,15 +266,15 @@ object TaskConversions extends Logging {
val metrics = BDPJettyServerHelper.gson.fromJson((job.getMetrics), classOf[util.Map[String, Object]])
var completeTime: Date = null
- if(null != metrics && metrics.containsKey(TaskConstant.ENTRANCEJOB_COMPLETE_TIME) && metrics.get(TaskConstant.ENTRANCEJOB_COMPLETE_TIME) != null){
+ if (null != metrics && metrics.containsKey(TaskConstant.ENTRANCEJOB_COMPLETE_TIME) && metrics.get(TaskConstant.ENTRANCEJOB_COMPLETE_TIME) != null) {
completeTime = dealString2Date(metrics.get(TaskConstant.ENTRANCEJOB_COMPLETE_TIME).toString)
}
var createTime: Date = null
- if(null != metrics && metrics.containsKey(TaskConstant.ENTRANCEJOB_SUBMIT_TIME) && metrics.get(TaskConstant.ENTRANCEJOB_SUBMIT_TIME) != null){
+ if (null != metrics && metrics.containsKey(TaskConstant.ENTRANCEJOB_SUBMIT_TIME) && metrics.get(TaskConstant.ENTRANCEJOB_SUBMIT_TIME) != null) {
createTime = dealString2Date(metrics.get(TaskConstant.ENTRANCEJOB_SUBMIT_TIME).toString)
}
if (null != createTime) {
- if(isJobFinished(job.getStatus)) {
+ if (isJobFinished(job.getStatus)) {
if (null != completeTime) {
taskVO.setCostTime(completeTime.getTime - createTime.getTime)
} else if (null != job.getUpdatedTime) {
@@ -286,6 +286,14 @@ object TaskConversions extends Logging {
taskVO.setCostTime(System.currentTimeMillis() - createTime.getTime)
}
}
+ if (metrics.containsKey(TaskConstant.ENTRANCEJOB_ENGINECONN_MAP)) {
+ val engineMap = metrics.get(TaskConstant.ENTRANCEJOB_ENGINECONN_MAP).asInstanceOf[util.Map[String, Object]]
+ if (null != engineMap && !engineMap.isEmpty) {
+ taskVO.setEngineInstance(engineMap.map(_._1).toList.mkString(","))
+ }
+ } else {
+ taskVO.setEngineInstance("EngineInstance not ready in metrics.")
+ }
val entranceName = JobhistoryConfiguration.ENTRANCE_SPRING_NAME.getValue
val instances = job.getInstances().split(JobhistoryConfiguration.ENTRANCE_INSTANCE_DELEMITER.getValue)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[incubator-linkis] 08/09: rename TaskConstants to TaskInfoConstants
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch dev-1.1.3
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
commit 229cb020a89afb92ce5c98b6982ebea8b781f795
Author: alexkun <xu...@qq.com>
AuthorDate: Tue Jun 7 17:51:47 2022 +0800
rename TaskConstants to TaskInfoConstants
---
.../job/{TaskConstants.scala => TaskInfoConstants.scala} | 2 +-
.../operation/progress/DefaultProgressOperation.scala | 10 +++++-----
2 files changed, 6 insertions(+), 6 deletions(-)
diff --git a/linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/constant/job/TaskConstants.scala b/linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/constant/job/TaskInfoConstants.scala
similarity index 97%
rename from linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/constant/job/TaskConstants.scala
rename to linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/constant/job/TaskInfoConstants.scala
index 1a6b50ec6..5c7902623 100644
--- a/linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/constant/job/TaskConstants.scala
+++ b/linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/constant/job/TaskInfoConstants.scala
@@ -17,7 +17,7 @@
package org.apache.linkis.governance.common.constant.job
-object TaskConstants {
+object TaskInfoConstants {
val EXEC_ID = "execId"
val ID_INFO = "idInfo"
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
index 1a900c053..f812976e9 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
@@ -17,7 +17,7 @@
package org.apache.linkis.orchestrator.computation.operation.progress
-import org.apache.linkis.governance.common.constant.job.TaskConstants
+import org.apache.linkis.governance.common.constant.job.TaskInfoConstants
import org.apache.linkis.orchestrator.OrchestratorSession
import org.apache.linkis.orchestrator.computation.physical.CodeLogicalUnitExecTask
import org.apache.linkis.orchestrator.computation.utils.TaskTreeUtil
@@ -113,10 +113,10 @@ class DefaultProgressOperation(orchestratorSession: OrchestratorSession) extends
logger.warn("TaskRunningInfoEvent got null execTask.")
return map
}
- map.put(TaskConstants.EXEC_ID, task.getId)
- map.put(TaskConstants.ID_INFO, task.getIDInfo())
- map.put(TaskConstants.TASK_CLASSNAME, task.getClass.getSimpleName)
- map.put(TaskConstants.TASK_NAME, task.getName)
+ map.put(TaskInfoConstants.EXEC_ID, task.getId)
+ map.put(TaskInfoConstants.ID_INFO, task.getIDInfo())
+ map.put(TaskInfoConstants.TASK_CLASSNAME, task.getClass.getSimpleName)
+ map.put(TaskInfoConstants.TASK_NAME, task.getName)
map
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[incubator-linkis] 01/09: optimize TaskYarnResourceEvent and TaskProgressEvent as one Event, defined as TaskRunningInfoEvent: 1. Record resource information 2. Record progress information 3. Record ec information
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch dev-1.1.3
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
commit 5c1ef13ce14ba630cbf4d9a762f1112f18acd8be
Author: alexkun <xu...@qq.com>
AuthorDate: Thu May 12 19:36:03 2022 +0800
optimize TaskYarnResourceEvent and TaskProgressEvent as one Event, defined as TaskRunningInfoEvent:
1. Record resource information
2. Record progress information
3. Record ec information
---
.../linkis/protocol/constants/TaskConstant.java | 3 ++
.../common/constant/job/TaskConstants.scala | 19 ++++----
.../service/TaskExecutionServiceImpl.scala | 25 +++++++++--
.../entrance/execute/DefaultEntranceExecutor.scala | 52 +++++++++++++---------
.../linkis/entrance/execute/EntranceExecutor.scala | 8 +---
.../EntranceOrchestrationFactory.scala | 1 -
.../protocol/resource/ResourceProtocol.scala | 4 ++
.../progress/AbstractProgressOperation.scala | 14 ++----
.../progress/DefaultProgressOperation.scala | 51 ++++++++++++++++-----
.../operation/progress/ProgressInfoEvent.scala | 6 ++-
.../operation/progress/ProgressProcessor.scala | 13 +++---
.../service/ComputationTaskExecutionReceiver.scala | 19 +++-----
.../orchestrator/listener/task/TaskInfoEvent.scala | 3 +-
.../listener/task/TaskProgressListener.scala | 2 +-
.../plans/logical/LogicalContext.scala | 4 +-
.../plans/logical/LogicalContextImpl.scala | 4 +-
.../plans/physical/PhysicalContext.scala | 6 +--
.../plans/physical/PhysicalContextImpl.scala | 9 ++--
.../ecm/service/TaskExecutionReceiver.scala | 6 +--
19 files changed, 142 insertions(+), 107 deletions(-)
diff --git a/linkis-commons/linkis-protocol/src/main/java/org/apache/linkis/protocol/constants/TaskConstant.java b/linkis-commons/linkis-protocol/src/main/java/org/apache/linkis/protocol/constants/TaskConstant.java
index f5cca82be..5d4b74960 100644
--- a/linkis-commons/linkis-protocol/src/main/java/org/apache/linkis/protocol/constants/TaskConstant.java
+++ b/linkis-commons/linkis-protocol/src/main/java/org/apache/linkis/protocol/constants/TaskConstant.java
@@ -61,6 +61,9 @@ public interface TaskConstant {
String ENTRANCEJOB_CORE_RGB = "coreRGB";
String ENTRANCEJOB_MEMORY_RGB = "memoryRGB";
+ String ENTRANCEJOB_ENGINECONN_MAP = "engineconnMap";
+ String ENGINE_INSTANCE = "engineInstance";
+
String PARAMS_DATA_SOURCE = "dataSources";
String PARAMS_CONTEXT = "context";
diff --git a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/listener/task/TaskProgressListener.scala b/linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/constant/job/TaskConstants.scala
similarity index 72%
copy from linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/listener/task/TaskProgressListener.scala
copy to linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/constant/job/TaskConstants.scala
index 4bdb66242..1a6b50ec6 100644
--- a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/listener/task/TaskProgressListener.scala
+++ b/linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/constant/job/TaskConstants.scala
@@ -5,24 +5,23 @@
* 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.orchestrator.listener.task
-import org.apache.linkis.orchestrator.listener.OrchestratorAsyncListener
+package org.apache.linkis.governance.common.constant.job
-/**
- * Progress Async Listener
- */
-trait TaskProgressListener extends OrchestratorAsyncListener {
+object TaskConstants {
+
+ val EXEC_ID = "execId"
+ val ID_INFO = "idInfo"
+ val TASK_CLASSNAME = "taskClassname"
+ val TASK_NAME = "taskName"
- def onProgressOn(taskProgressEvent: TaskProgressEvent): Unit
}
diff --git a/linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/service/TaskExecutionServiceImpl.scala b/linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/service/TaskExecutionServiceImpl.scala
index 6e927867f..420f95645 100644
--- a/linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/service/TaskExecutionServiceImpl.scala
+++ b/linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/service/TaskExecutionServiceImpl.scala
@@ -44,8 +44,9 @@ import org.apache.linkis.governance.common.exception.engineconn.{EngineConnExecu
import org.apache.linkis.governance.common.protocol.task._
import org.apache.linkis.governance.common.utils.JobUtils
import org.apache.linkis.manager.common.entity.enumeration.NodeStatus
-import org.apache.linkis.manager.common.protocol.resource.ResponseTaskYarnResource
+import org.apache.linkis.manager.common.protocol.resource.{ResponseTaskRunningInfo, ResponseTaskYarnResource}
import org.apache.linkis.manager.label.entity.Label
+import org.apache.linkis.protocol.constants.TaskConstant
import org.apache.linkis.protocol.message.RequestProtocol
import org.apache.linkis.rpc.Sender
import org.apache.linkis.rpc.message.annotation.Receiver
@@ -55,6 +56,7 @@ import org.apache.linkis.server.BDPJettyServerHelper
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Component
+import java.util
import java.util.concurrent._
import java.util.concurrent.atomic.AtomicInteger
import javax.annotation.PostConstruct
@@ -319,15 +321,30 @@ class TaskExecutionServiceImpl extends TaskExecutionService with Logging with Re
Utils.tryQuietly(Thread.sleep(TimeUnit.MILLISECONDS.convert(1, TimeUnit.SECONDS)))
while (null != taskFuture && !taskFuture.isDone) {
if (ExecutionNodeStatus.isCompleted(task.getStatus) || ExecutionNodeStatus.isRunning(task.getStatus)) {
- sendToEntrance(task, taskProgress(task.getTaskId))
- taskYarnResource(task.getTaskId) match {
+ val progressResponse = taskProgress(task.getTaskId)
+ val resourceResponse: ResponseTaskYarnResource = taskYarnResource(task.getTaskId) match {
case responseTaskYarnResource: ResponseTaskYarnResource => {
if (responseTaskYarnResource.resourceMap != null && !responseTaskYarnResource.resourceMap.isEmpty) {
- sendToEntrance(task, responseTaskYarnResource)
+ responseTaskYarnResource
+ } else {
+ null
}
}
case _ =>
+ null
}
+ val extraInfoMap = new util.HashMap[String, Object]()
+ extraInfoMap.put(TaskConstant.ENGINE_INSTANCE, Sender.getThisInstance)
+ // todo add other info
+ var respRunningInfo: ResponseTaskRunningInfo = null
+ if (null != resourceResponse) {
+ respRunningInfo = ResponseTaskRunningInfo(progressResponse.execId, progressResponse.progress, progressResponse.progressInfo,
+ resourceResponse.resourceMap, extraInfoMap)
+ } else {
+ respRunningInfo = ResponseTaskRunningInfo(progressResponse.execId, progressResponse.progress, progressResponse.progressInfo,
+ null, extraInfoMap)
+ }
+ sendToEntrance(task, respRunningInfo)
Thread.sleep(TimeUnit.MILLISECONDS.convert(sleepInterval, TimeUnit.SECONDS))
}
}
diff --git a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/execute/DefaultEntranceExecutor.scala b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/execute/DefaultEntranceExecutor.scala
index cd5540dd3..c6165523c 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/execute/DefaultEntranceExecutor.scala
+++ b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/execute/DefaultEntranceExecutor.scala
@@ -75,23 +75,7 @@ class DefaultEntranceExecutor(id: Long, mark: MarkReq, entranceExecutorManager:
logProcessor
}
- def dealResourceReport(orchestratorFuture: OrchestrationFuture, job: EntranceJob): ResourceReportProcessor = {
- val resourceReportProcessor = orchestratorFuture.operate[ResourceReportProcessor](ResourceReportOperation.RESOURCE)
- resourceReportProcessor.registerResourceReportNotify(resourceReportEvent => {
- if (job.getJobRequest.getMetrics == null) {
- job.getJobRequest.setMetrics(new util.HashMap[String, Object]())
- }
- val resourceMap = job.getJobRequest.getMetrics.get(TaskConstant.ENTRANCEJOB_YARNRESOURCE)
- if(resourceMap != null) {
- resourceMap.asInstanceOf[util.HashMap[String, ResourceWithStatus]].putAll(resourceReportEvent.resourceMap)
- } else {
- job.getJobRequest.getMetrics.put(TaskConstant.ENTRANCEJOB_YARNRESOURCE, resourceReportEvent.resourceMap)
- }
- })
- resourceReportProcessor
- }
-
- def dealProgress(orchestratorFuture: OrchestrationFuture, entranceJob: EntranceJob): ProgressProcessor = {
+ def dealProgressWithResource(orchestratorFuture: OrchestrationFuture, entranceJob: EntranceJob): ProgressProcessor = {
val progressProcessor = orchestratorFuture.operate[ProgressProcessor](DefaultProgressOperation.PROGRESS_NAME)
progressProcessor.doOnObtain(progressInfoEvent => {
if (null != entranceJob) {
@@ -118,6 +102,35 @@ class DefaultEntranceExecutor(id: Long, mark: MarkReq, entranceExecutorManager:
entranceJob.getProgressListener.foreach(_.onProgressUpdate(entranceJob, progressInfoEvent.progress,
entranceJob.getProgressInfo))
}
+ // update resource
+ if (entranceJob.getJobRequest.getMetrics == null) {
+ entranceJob.getJobRequest.setMetrics(new util.HashMap[String, Object]())
+ }
+ val metricsMap = entranceJob.getJobRequest.getMetrics
+ val resourceMap = metricsMap.get(TaskConstant.ENTRANCEJOB_YARNRESOURCE)
+ if(resourceMap != null) {
+ resourceMap.asInstanceOf[util.HashMap[String, ResourceWithStatus]].putAll(progressInfoEvent.resourceMap)
+ } else {
+ metricsMap.put(TaskConstant.ENTRANCEJOB_YARNRESOURCE, progressInfoEvent.resourceMap)
+ }
+ // update engine info
+ // todo
+ var engineInstanceMap: util.HashMap[String, Object] = null
+ if (metricsMap.containsKey(TaskConstant.ENTRANCEJOB_ENGINECONN_MAP)) {
+ engineInstanceMap = metricsMap.get(TaskConstant.ENTRANCEJOB_ENGINECONN_MAP).asInstanceOf[util.HashMap[String, Object]]
+ } else {
+ engineInstanceMap = new util.HashMap[String, Object]()
+ metricsMap.put(TaskConstant.ENTRANCEJOB_ENGINECONN_MAP, engineInstanceMap)
+ }
+ val infoMap = progressInfoEvent.infoMap
+ if (infoMap.containsKey(TaskConstant.ENGINE_INSTANCE)) {
+ val instance = infoMap.get(TaskConstant.ENGINE_INSTANCE).asInstanceOf[String]
+ val engineExtraInfoMap = engineInstanceMap.getOrDefault(instance, new util.HashMap[String, Object]).asInstanceOf[util.HashMap[String, Object]]
+ engineInstanceMap.put(instance, engineExtraInfoMap)
+ engineExtraInfoMap.putAll(infoMap)
+ } else {
+ logger.warn("Engine extra info map must contains engineInstance")
+ }
}
})
progressProcessor
@@ -288,8 +301,7 @@ class DefaultEntranceExecutor(id: Long, mark: MarkReq, entranceExecutorManager:
}
// 2. deal log And Response
val logProcessor = dealLog(orchestratorFuture, entranceExecuteRequest.getJob)
- val progressProcessor = dealProgress(orchestratorFuture, entranceExecuteRequest.getJob)
- val resourceReportProcessor = dealResourceReport(orchestratorFuture, entranceExecuteRequest.getJob)
+ val progressAndResourceProcessor = dealProgressWithResource(orchestratorFuture, entranceExecuteRequest.getJob)
orchestratorFuture.notifyMe(orchestrationResponse => {
dealResponse(orchestrationResponse, entranceExecuteRequest, orchestration)
}
@@ -302,7 +314,7 @@ class DefaultEntranceExecutor(id: Long, mark: MarkReq, entranceExecutorManager:
logger.info(s"For job ${entranceExecuteRequest.getJob.getId} and subJob ${compJobReq.getId} to create EngineExecuteAsyncReturn")
new EngineExecuteAsyncReturn(request, null)
}
- jobReturn.setOrchestrationObjects(orchestratorFuture, logProcessor, progressProcessor, resourceReportProcessor)
+ jobReturn.setOrchestrationObjects(orchestratorFuture, logProcessor, progressAndResourceProcessor)
jobReturn.setSubJobId(compJobReq.getId)
setEngineReturn(jobReturn)
jobReturn
diff --git a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/execute/EntranceExecutor.scala b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/execute/EntranceExecutor.scala
index 529e8d24e..f78146860 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/execute/EntranceExecutor.scala
+++ b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/execute/EntranceExecutor.scala
@@ -141,11 +141,10 @@ class EngineExecuteAsyncReturn(val request: ExecuteRequest,
def getLastNotifyTime: Long = lastNotifyTime
- def setOrchestrationObjects(orchestrationFuture: OrchestrationFuture, logProcessor: LogProcessor, progressProcessor: ProgressProcessor, resourceReportProcessor: ResourceReportProcessor): Unit = {
+ def setOrchestrationObjects(orchestrationFuture: OrchestrationFuture, logProcessor: LogProcessor, progressProcessor: ProgressProcessor): Unit = {
this.orchestrationFuture = orchestrationFuture
this.logProcessor = logProcessor
this.progressProcessor = progressProcessor
- this.resourceReportProcessor = resourceReportProcessor
}
def getOrchestrationFuture(): Option[OrchestrationFuture] = {
@@ -160,14 +159,9 @@ class EngineExecuteAsyncReturn(val request: ExecuteRequest,
Option(progressProcessor)
}
- def getResourceReportProcessor(): Option[ResourceReportProcessor] = {
- Option(resourceReportProcessor)
- }
-
def closeOrchestration(): Unit = {
getLogProcessor().foreach(IOUtils.closeQuietly(_))
getProgressProcessor().foreach(IOUtils.closeQuietly(_))
- getResourceReportProcessor().foreach(IOUtils.closeQuietly(_))
}
def setSubJobId(subJobId: String): Unit = {
diff --git a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/orchestrator/EntranceOrchestrationFactory.scala b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/orchestrator/EntranceOrchestrationFactory.scala
index 1db6945b2..0a53542f4 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/orchestrator/EntranceOrchestrationFactory.scala
+++ b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/orchestrator/EntranceOrchestrationFactory.scala
@@ -37,7 +37,6 @@ object EntranceOrchestrationFactory {
val addOnOperation = new OperationExtensionsBuilder {
override def apply(v1: OperationExtensions): Unit = {
v1.injectOperation(new ProgressOperationBuilder())
- v1.injectOperation(new ResourceReportOperationBuilder())
}
}
orchestratorSessionBuilder.withOperationExtensions(addOnOperation)
diff --git a/linkis-computation-governance/linkis-manager/linkis-manager-commons/linkis-manager-common/src/main/scala/org/apache/linkis/manager/common/protocol/resource/ResourceProtocol.scala b/linkis-computation-governance/linkis-manager/linkis-manager-commons/linkis-manager-common/src/main/scala/org/apache/linkis/manager/common/protocol/resource/ResourceProtocol.scala
index a505cbdf7..3a04fac4a 100644
--- a/linkis-computation-governance/linkis-manager/linkis-manager-commons/linkis-manager-common/src/main/scala/org/apache/linkis/manager/common/protocol/resource/ResourceProtocol.scala
+++ b/linkis-computation-governance/linkis-manager/linkis-manager-commons/linkis-manager-common/src/main/scala/org/apache/linkis/manager/common/protocol/resource/ResourceProtocol.scala
@@ -19,6 +19,8 @@ package org.apache.linkis.manager.common.protocol.resource
import org.apache.linkis.common.ServiceInstance
import org.apache.linkis.manager.common.entity.resource.NodeResource
+import org.apache.linkis.protocol.RetryableProtocol
+import org.apache.linkis.protocol.engine.JobProgressInfo
import org.apache.linkis.protocol.message.RequestProtocol
import java.util
@@ -29,3 +31,5 @@ case class ResourceUsedProtocol(serviceInstance: ServiceInstance, engineResource
case class ResponseTaskYarnResource(execId: String, resourceMap: util.HashMap[String, ResourceWithStatus]) extends RequestProtocol
+case class ResponseTaskRunningInfo(execId: String, progress: Float, progressInfo: Array[JobProgressInfo],
+ resourceMap: util.HashMap[String, ResourceWithStatus], extraInfoMap: util.HashMap[String, Object]) extends RetryableProtocol with RequestProtocol
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/AbstractProgressOperation.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/AbstractProgressOperation.scala
index 0ee652657..f1c6d5f98 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/AbstractProgressOperation.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/AbstractProgressOperation.scala
@@ -24,7 +24,7 @@ import org.apache.linkis.orchestrator.computation.operation.progress.ProgressPro
import org.apache.linkis.orchestrator.core.AbstractOrchestration
import org.apache.linkis.orchestrator.extensions.operation.Operation
import org.apache.linkis.orchestrator.extensions.operation.Operation.OperationBuilder
-import org.apache.linkis.orchestrator.listener.task.{TaskProgressEvent, TaskProgressListener}
+import org.apache.linkis.orchestrator.listener.task.{TaskRunningInfoEvent, TaskProgressListener}
import org.apache.linkis.orchestrator.listener.{OrchestratorAsyncEvent, OrchestratorListenerBusContext}
import org.apache.linkis.orchestrator.{Orchestration, OrchestratorSession}
@@ -55,8 +55,8 @@ abstract class AbstractProgressOperation(orchestratorSession: OrchestratorSessio
override def onEvent(event: OrchestratorAsyncEvent): Unit = {
event match {
- case progressEvent: TaskProgressEvent =>
- onProgressOn(progressEvent)
+ case progressWithResourceEvent: TaskRunningInfoEvent =>
+ onProgressOn(progressWithResourceEvent)
case _ =>
}
}
@@ -65,14 +65,6 @@ abstract class AbstractProgressOperation(orchestratorSession: OrchestratorSessio
this.execTaskToProgressProcessor.remove(execTaskId)
}
- override def onProgressOn(taskProgressEvent: TaskProgressEvent): Unit = {
- /*val execTask = taskProgressEvent.execTask
- Option(execTaskToProgressProcessor.get(execTask.getPhysicalContext.getRootTask.getId)).foreach( progress => {
- progress.onProgress(taskProgressEvent.progress,
- taskProgressEvent.progressInfo)
- })*/
- warn("Should not be called.")
- }
override def onEventError(event: Event, t: Throwable): Unit = {
var eventName: String = "Null Event"
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
index 1d34abe75..88f5fb90a 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
@@ -17,11 +17,15 @@
package org.apache.linkis.orchestrator.computation.operation.progress
+import org.apache.linkis.governance.common.constant.job.TaskConstants
import org.apache.linkis.orchestrator.OrchestratorSession
import org.apache.linkis.orchestrator.computation.physical.CodeLogicalUnitExecTask
import org.apache.linkis.orchestrator.computation.utils.TaskTreeUtil
-import org.apache.linkis.orchestrator.listener.task.TaskProgressEvent
+import org.apache.linkis.orchestrator.listener.task.TaskRunningInfoEvent
+import org.apache.linkis.orchestrator.plans.physical.ExecTask
+import org.apache.linkis.protocol.constants.TaskConstant
+import java.util
import java.util.concurrent.ConcurrentHashMap
import java.util.function.{DoubleBinaryOperator, ToDoubleFunction}
import scala.collection.JavaConverters.asScalaBufferConverter
@@ -51,46 +55,69 @@ class DefaultProgressOperation(orchestratorSession: OrchestratorSession) extends
/**
* To deal with the progress event from engineConn
- * @param taskProgressEvent progress event
+ * @param taskRunningInfoEvent progress event
*/
- override def onProgressOn(taskProgressEvent: TaskProgressEvent): Unit = {
- val execTask = taskProgressEvent.execTask
+ override def onProgressOn(oriTaskRunningInfoEvent: TaskRunningInfoEvent): Unit = {
+ var taskRunningInfoEvent: TaskRunningInfoEvent = null
+ val execTaskInfo = generateExecTaskInfo(oriTaskRunningInfoEvent.execTask)
+ if (null != taskRunningInfoEvent.infoMap) {
+ taskRunningInfoEvent = TaskRunningInfoEvent(oriTaskRunningInfoEvent.execTask, oriTaskRunningInfoEvent.progress,
+ oriTaskRunningInfoEvent.progressInfo, oriTaskRunningInfoEvent.resourceMap, ta)
+ } else {
+ taskRunningInfoEvent = TaskRunningInfoEvent(oriTaskRunningInfoEvent.execTask, oriTaskRunningInfoEvent.progress,
+ oriTaskRunningInfoEvent.progressInfo, oriTaskRunningInfoEvent.resourceMap, generateExecTaskInfo(oriTaskRunningInfoEvent.execTask))
+ }
+ val execTask = taskRunningInfoEvent.execTask
Option(execTask).foreach(task => {
val physicalContext = task.getPhysicalContext
val progressInfo = physicalContext.getOption(ProgressConstraints.PROGRESS_MAP_NAME)
- val event = if(null != progressInfo.get){
+ val event = if (null != progressInfo.get) {
val progressMap = progressInfo.get.asInstanceOf[ConcurrentHashMap[String, Float]]
//Update the progress value
- progressMap.put(execTask.getId, taskProgressEvent.progress)
+ progressMap.put(execTask.getId, taskRunningInfoEvent.progress)
//Iterate the progress map and calculate the global progress value
progressMap synchronized {
//Make sure that the iterate function is serial
- val progressValueSum = progressMap.reduceValuesToDouble(1L,new ToDoubleFunction[Float] {
+ val progressValueSum = progressMap.reduceValuesToDouble(1L, new ToDoubleFunction[Float] {
override def applyAsDouble(t: Float): Double = t.asInstanceOf[Double]
}, 0.0d, new DoubleBinaryOperator {
override def applyAsDouble(left: Double, right: Double): Double = left + right
})
//Update the global progress value in event
- TaskProgressEvent(execTask, progressValueSum.asInstanceOf[Float] / progressMap.size.asInstanceOf[Float], taskProgressEvent.progressInfo)
+ TaskRunningInfoEvent(execTask, progressValueSum.asInstanceOf[Float] / progressMap.size.asInstanceOf[Float],
+ taskRunningInfoEvent.progressInfo, taskRunningInfoEvent.resourceMap, taskRunningInfoEvent.infoMap)
}
} else {
val progressMap = physicalContext.getOrElsePut(ProgressConstraints.PROGRESS_MAP_NAME, new ConcurrentHashMap[String, Float])
- .asInstanceOf[ConcurrentHashMap[String,Float]]
+ .asInstanceOf[ConcurrentHashMap[String, Float]]
//Init the value of progress as 0
val codeExecTasks = TaskTreeUtil.getAllTaskRecursive(physicalContext.getRootTask, classOf[CodeLogicalUnitExecTask])
if (null != codeExecTasks) {
codeExecTasks.asScala.foreach(task => progressMap.put(task.getId, 0.0f))
}
- progressMap.put(execTask.getId, taskProgressEvent.progress)
- taskProgressEvent
+ progressMap.put(execTask.getId, taskRunningInfoEvent.progress)
+ taskRunningInfoEvent
}
Option(execTaskToProgressProcessor.get(execTask.getPhysicalContext.getRootTask.getId)).foreach( progress => {
- progress.onProgress(event.progress, event.progressInfo)
+ progress.onProgress(event.progress, event.progressInfo, event.resourceMap, event.resourceMap, event.infoMap)
})
})
}
+
+ private def generateExecTaskInfo(task: ExecTask): util.HashMap[String, Object] = {
+ val map = new util.HashMap[String, Object]()
+ if (null == task) {
+ logger.warn("TaskRunningInfoEvent got null execTask.")
+ return map
+ }
+ map.put(TaskConstants.EXEC_ID, task.getId())
+ map.put(TaskConstants.ID_INFO, task.getIDInfo())
+ map.put(TaskConstants.TASK_CLASSNAME, task.getClass.getSimpleName)
+ map.put(TaskConstants.TASK_NAME, task.getName)
+ map
+ }
}
object DefaultProgressOperation{
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressInfoEvent.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressInfoEvent.scala
index f634a3932..e2d86a046 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressInfoEvent.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressInfoEvent.scala
@@ -17,9 +17,13 @@
package org.apache.linkis.orchestrator.computation.operation.progress
+import org.apache.linkis.manager.common.protocol.resource.ResourceWithStatus
import org.apache.linkis.orchestrator.Orchestration
import org.apache.linkis.protocol.engine.JobProgressInfo
+import java.util
case class ProgressInfoEvent(orchestration: Orchestration,
- progress: Float, progressInfo: Array[JobProgressInfo])
+ progress: Float, progressInfo: Array[JobProgressInfo],
+ resourceMap: util.HashMap[String, ResourceWithStatus],
+ infoMap: util.HashMap[String, Object])
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressProcessor.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressProcessor.scala
index 7386e7b9d..680044651 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressProcessor.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressProcessor.scala
@@ -17,17 +17,16 @@
package org.apache.linkis.orchestrator.computation.operation.progress
-import java.io.Closeable
+import org.apache.linkis.manager.common.protocol.resource.ResourceWithStatus
+import java.io.Closeable
import org.apache.linkis.orchestrator.Orchestration
import org.apache.linkis.orchestrator.computation.operation
import org.apache.linkis.protocol.engine.JobProgressInfo
+import java.util
import scala.collection.mutable.ArrayBuffer
-/**
- *
- */
class ProgressProcessor(rootExecTaskId: String,
orchestration: Orchestration, progressObtainOperation: AbstractProgressOperation) extends Closeable{
@@ -41,9 +40,9 @@ class ProgressProcessor(rootExecTaskId: String,
listeners += notify
}
- def onProgress(progress: Float, progressInfo: Array[JobProgressInfo]): Unit = {
- val progressInfoEvent = operation.progress.ProgressInfoEvent(orchestration, progress, progressInfo)
- listeners.foreach(_(progressInfoEvent))
+ def onProgress(progress: Float, progressInfo: Array[JobProgressInfo], resourceMap: util.HashMap[String, ResourceWithStatus], infoMap: util.HashMap[String, Object]): Unit = {
+ val progressInfoWithResourceEvent = ProgressInfoEvent(orchestration, progress, progressInfo, resourceMap)
+ listeners.foreach(_(progressInfoWithResourceEvent))
}
override def close(): Unit = {
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/service/ComputationTaskExecutionReceiver.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/service/ComputationTaskExecutionReceiver.scala
index 0f342ecb8..2e7860f51 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/service/ComputationTaskExecutionReceiver.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/service/ComputationTaskExecutionReceiver.scala
@@ -21,7 +21,7 @@ import org.apache.linkis.common.log.LogUtils
import org.apache.linkis.common.utils.{Logging, Utils}
import org.apache.linkis.governance.common.entity.ExecutionNodeStatus
import org.apache.linkis.governance.common.protocol.task._
-import org.apache.linkis.manager.common.protocol.resource.ResponseTaskYarnResource
+import org.apache.linkis.manager.common.protocol.resource.{ResponseTaskRunningInfo, ResponseTaskYarnResource}
import org.apache.linkis.orchestrator.computation.execute.CodeExecTaskExecutorManager
import org.apache.linkis.orchestrator.computation.monitor.EngineConnMonitor
import org.apache.linkis.orchestrator.core.ResultSet
@@ -86,20 +86,11 @@ class ComputationTaskExecutionReceiver extends TaskExecutionReceiver with Loggin
}
@Receiver
- override def taskYarnResourceReceiver(taskYarnResource: ResponseTaskYarnResource, sender: Sender): Unit = {
+ override def taskProgressReceiver(taskProgressWithResource: ResponseTaskRunningInfo, sender: Sender): Unit = {
val serviceInstance = RPCUtils.getServiceInstanceFromSender(sender)
- codeExecTaskExecutorManager.getByEngineConnAndTaskId(serviceInstance, taskYarnResource.execId).foreach { codeExecutor =>
- val event = TaskYarnResourceEvent(codeExecutor.getExecTask, taskYarnResource.resourceMap)
- codeExecutor.getExecTask.getPhysicalContext.pushYarnResource(event)
- codeExecutor.getEngineConnExecutor.updateLastUpdateTime()
- }
- }
-
- @Receiver
- override def taskProgressReceiver(taskProgress: ResponseTaskProgress, sender: Sender): Unit = {
- val serviceInstance = RPCUtils.getServiceInstanceFromSender(sender)
- codeExecTaskExecutorManager.getByEngineConnAndTaskId(serviceInstance, taskProgress.execId).foreach{ codeExecutor =>
- val event = TaskProgressEvent(codeExecutor.getExecTask, taskProgress.progress, taskProgress.progressInfo)
+ codeExecTaskExecutorManager.getByEngineConnAndTaskId(serviceInstance, taskProgressWithResource.execId).foreach{ codeExecutor =>
+ val event = TaskRunningInfoEvent(codeExecutor.getExecTask, taskProgressWithResource.progress,
+ taskProgressWithResource.progressInfo, taskProgressWithResource.resourceMap, taskProgressWithResource.extraInfoMap)
codeExecutor.getExecTask.getPhysicalContext.pushProgress(event)
codeExecutor.getEngineConnExecutor.updateLastUpdateTime()
}
diff --git a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/listener/task/TaskInfoEvent.scala b/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/listener/task/TaskInfoEvent.scala
index d53ceed55..8cbe0ded6 100644
--- a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/listener/task/TaskInfoEvent.scala
+++ b/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/listener/task/TaskInfoEvent.scala
@@ -70,7 +70,8 @@ case class KillRootExecTaskEvent(execTask: ExecTask) extends TaskInfoEvent with
case class TaskReheaterEvent(execTask: ExecTask) extends TaskInfoEvent with OrchestratorAsyncEvent
-case class TaskProgressEvent(execTask: ExecTask, progress: Float, progressInfo: Array[JobProgressInfo]) extends TaskInfoEvent with OrchestratorAsyncEvent {
+case class TaskRunningInfoEvent(execTask: ExecTask, progress: Float, progressInfo: Array[JobProgressInfo],
+ resourceMap: util.HashMap[String, ResourceWithStatus], infoMap: util.HashMap[String, Object]) extends TaskInfoEvent with OrchestratorAsyncEvent {
override def toString: String = {
s"task ${execTask.getIDInfo()}, progress ${progress}"
}
diff --git a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/listener/task/TaskProgressListener.scala b/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/listener/task/TaskProgressListener.scala
index 4bdb66242..313aa75bb 100644
--- a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/listener/task/TaskProgressListener.scala
+++ b/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/listener/task/TaskProgressListener.scala
@@ -24,5 +24,5 @@ import org.apache.linkis.orchestrator.listener.OrchestratorAsyncListener
*/
trait TaskProgressListener extends OrchestratorAsyncListener {
- def onProgressOn(taskProgressEvent: TaskProgressEvent): Unit
+ def onProgressOn(taskProgressEvent: TaskRunningInfoEvent): Unit
}
diff --git a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/logical/LogicalContext.scala b/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/logical/LogicalContext.scala
index 439ee4d04..e82232308 100644
--- a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/logical/LogicalContext.scala
+++ b/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/logical/LogicalContext.scala
@@ -18,7 +18,7 @@
package org.apache.linkis.orchestrator.plans.logical
import org.apache.linkis.common.listener.Event
-import org.apache.linkis.orchestrator.listener.task.{TaskLogEvent, TaskProgressEvent}
+import org.apache.linkis.orchestrator.listener.task.{TaskLogEvent, TaskRunningInfoEvent}
import org.apache.linkis.orchestrator.plans.PlanContext
/**
@@ -38,7 +38,7 @@ trait LogicalContext extends PlanContext {
def pushLog(taskLogEvent: TaskLogEvent): Unit
- def pushProgress(taskProgressEvent: TaskProgressEvent): Unit
+ def pushProgress(taskProgressEvent: TaskRunningInfoEvent): Unit
def broadcastAsyncEvent(event: Event): Unit
diff --git a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/logical/LogicalContextImpl.scala b/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/logical/LogicalContextImpl.scala
index fe465cbc7..d8a4ce71b 100644
--- a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/logical/LogicalContextImpl.scala
+++ b/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/logical/LogicalContextImpl.scala
@@ -20,7 +20,7 @@ package org.apache.linkis.orchestrator.plans.logical
import java.util
import org.apache.linkis.common.listener.Event
-import org.apache.linkis.orchestrator.listener.task.{TaskLogEvent, TaskProgressEvent}
+import org.apache.linkis.orchestrator.listener.task.{TaskLogEvent, TaskRunningInfoEvent}
import scala.collection.mutable.ArrayBuffer
@@ -95,7 +95,7 @@ class LogicalContextImpl extends LogicalContext{
override def pushLog(taskLogEvent: TaskLogEvent): Unit = {}
- override def pushProgress(taskProgressEvent: TaskProgressEvent): Unit = {}
+ override def pushProgress(taskProgressEvent: TaskRunningInfoEvent): Unit = {}
override def broadcastAsyncEvent(event: Event): Unit = {}
diff --git a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/physical/PhysicalContext.scala b/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/physical/PhysicalContext.scala
index 8b0b199a1..3e0e7a8fe 100644
--- a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/physical/PhysicalContext.scala
+++ b/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/physical/PhysicalContext.scala
@@ -19,7 +19,7 @@ package org.apache.linkis.orchestrator.plans.physical
import org.apache.linkis.common.listener.Event
import org.apache.linkis.orchestrator.execution.TaskResponse
-import org.apache.linkis.orchestrator.listener.task.{TaskLogEvent, TaskProgressEvent, TaskYarnResourceEvent}
+import org.apache.linkis.orchestrator.listener.task.{TaskLogEvent, TaskRunningInfoEvent, TaskYarnResourceEvent}
import org.apache.linkis.orchestrator.plans.PlanContext
/**
@@ -41,9 +41,7 @@ trait PhysicalContext extends PlanContext {
def pushLog(taskLogEvent: TaskLogEvent): Unit
- def pushProgress(taskProgressEvent: TaskProgressEvent): Unit
-
- def pushYarnResource(taskProgressEvent: TaskYarnResourceEvent): Unit
+ def pushProgress(taskProgressEvent: TaskRunningInfoEvent): Unit
def belongsTo(execTask: ExecTask): Boolean
diff --git a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/physical/PhysicalContextImpl.scala b/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/physical/PhysicalContextImpl.scala
index b28acb3bf..fae703baf 100644
--- a/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/physical/PhysicalContextImpl.scala
+++ b/linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/physical/PhysicalContextImpl.scala
@@ -24,7 +24,7 @@ import org.apache.linkis.orchestrator.exception.OrchestratorErrorCodeSummary
import org.apache.linkis.orchestrator.execution.impl.DefaultFailedTaskResponse
import org.apache.linkis.orchestrator.execution.{CompletedTaskResponse, SucceedTaskResponse, TaskResponse}
import org.apache.linkis.orchestrator.listener._
-import org.apache.linkis.orchestrator.listener.task.{RootTaskResponseEvent, TaskLogEvent, TaskProgressEvent, TaskYarnResourceEvent}
+import org.apache.linkis.orchestrator.listener.task.{RootTaskResponseEvent, TaskLogEvent, TaskRunningInfoEvent, TaskYarnResourceEvent}
import scala.collection.mutable
import scala.collection.mutable.ListBuffer
@@ -175,11 +175,8 @@ class PhysicalContextImpl(private var rootTask: ExecTask, private var leafTasks:
broadcastAsyncEvent(taskLogEvent)
}
- override def pushProgress(taskProgressEvent: TaskProgressEvent): Unit = {
- broadcastAsyncEvent(taskProgressEvent)
- }
- override def pushYarnResource(taskYarnResourceEvent: TaskYarnResourceEvent): Unit = {
- broadcastAsyncEvent(taskYarnResourceEvent)
+ override def pushProgress(taskRunningInfoEvent: TaskRunningInfoEvent): Unit = {
+ broadcastAsyncEvent(taskRunningInfoEvent)
}
def setAsyncBus(asyncListenerBus: OrchestratorAsyncListenerBus): Unit = {
diff --git a/linkis-orchestrator/plugin/linkis-orchestrator-ecm-plugin/src/main/scala/org/apache/linkis/orchestrator/ecm/service/TaskExecutionReceiver.scala b/linkis-orchestrator/plugin/linkis-orchestrator-ecm-plugin/src/main/scala/org/apache/linkis/orchestrator/ecm/service/TaskExecutionReceiver.scala
index ff6a711e2..3cf4dbead 100644
--- a/linkis-orchestrator/plugin/linkis-orchestrator-ecm-plugin/src/main/scala/org/apache/linkis/orchestrator/ecm/service/TaskExecutionReceiver.scala
+++ b/linkis-orchestrator/plugin/linkis-orchestrator-ecm-plugin/src/main/scala/org/apache/linkis/orchestrator/ecm/service/TaskExecutionReceiver.scala
@@ -18,7 +18,7 @@
package org.apache.linkis.orchestrator.ecm.service
import org.apache.linkis.governance.common.protocol.task._
-import org.apache.linkis.manager.common.protocol.resource.ResponseTaskYarnResource
+import org.apache.linkis.manager.common.protocol.resource.{ResponseTaskRunningInfo, ResponseTaskYarnResource}
import org.apache.linkis.rpc.Sender
@@ -28,9 +28,7 @@ trait TaskExecutionReceiver {
def taskLogReceiver(taskLog: ResponseTaskLog, sender: Sender): Unit
- def taskProgressReceiver(taskProgress: ResponseTaskProgress, sender: Sender): Unit
-
- def taskYarnResourceReceiver(taskYarnResource: ResponseTaskYarnResource, sender: Sender): Unit
+ def taskProgressReceiver(taskProgress: ResponseTaskRunningInfo, sender: Sender): Unit
def taskStatusReceiver(taskStatus: ResponseTaskStatus, sender: Sender): Unit
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[incubator-linkis] 03/09: 1. fix DefaultProgressOperation
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch dev-1.1.3
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
commit 27f91caea96c9bb52a05ea6fb35ec7cceb3c3540
Author: alexkun <xu...@qq.com>
AuthorDate: Mon May 16 21:30:20 2022 +0800
1. fix DefaultProgressOperation
---
.../computation/operation/progress/AbstractProgressOperation.scala | 4 ++--
.../computation/operation/progress/DefaultProgressOperation.scala | 6 +++---
2 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/AbstractProgressOperation.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/AbstractProgressOperation.scala
index f1c6d5f98..5840e684b 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/AbstractProgressOperation.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/AbstractProgressOperation.scala
@@ -74,12 +74,12 @@ abstract class AbstractProgressOperation(orchestratorSession: OrchestratorSessio
eventName = event.getClass.getName
}
if (null != t) {
- message = t.getMessage
+ message = t.getClass.getSimpleName + t.getMessage
if (null != t.getCause) {
cause = t.getCause.getMessage
}
}
- debug(s"Accept error event ${eventName} in progress operation, message: ${message}, cause : ${cause}")
+ logger.warn(s"Accept error event ${eventName} in progress operation, message: ${message}, cause : ${cause}")
}
}
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
index 4f6819cc7..1a900c053 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
@@ -60,10 +60,10 @@ class DefaultProgressOperation(orchestratorSession: OrchestratorSession) extends
override def onProgressOn(oriTaskRunningInfoEvent: TaskRunningInfoEvent): Unit = {
var taskRunningInfoEvent: TaskRunningInfoEvent = null
val execTaskInfo = generateExecTaskInfo(oriTaskRunningInfoEvent.execTask)
- if (null != taskRunningInfoEvent.infoMap) {
- taskRunningInfoEvent.infoMap.putAll(execTaskInfo)
+ if (null != oriTaskRunningInfoEvent.infoMap) {
+ oriTaskRunningInfoEvent.infoMap.putAll(execTaskInfo)
taskRunningInfoEvent = TaskRunningInfoEvent(oriTaskRunningInfoEvent.execTask, oriTaskRunningInfoEvent.progress,
- oriTaskRunningInfoEvent.progressInfo, oriTaskRunningInfoEvent.resourceMap, taskRunningInfoEvent.infoMap)
+ oriTaskRunningInfoEvent.progressInfo, oriTaskRunningInfoEvent.resourceMap, oriTaskRunningInfoEvent.infoMap)
} else {
taskRunningInfoEvent = TaskRunningInfoEvent(oriTaskRunningInfoEvent.execTask, oriTaskRunningInfoEvent.progress,
oriTaskRunningInfoEvent.progressInfo, oriTaskRunningInfoEvent.resourceMap, execTaskInfo)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[incubator-linkis] 09/09: Fix NPE
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch dev-1.1.3
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
commit 16f645bb840ed6d48f9ae9fa61c0f76cbd945f52
Author: peacewong <wp...@gmail.com>
AuthorDate: Tue Jun 7 20:00:37 2022 +0800
Fix NPE
---
.../linkis/entrance/restful/EntranceRestfulApi.java | 2 +-
.../linkis/entrance/execute/DefaultEntranceExecutor.scala | 15 +++++++++------
.../operation/progress/DefaultProgressOperation.scala | 2 +-
.../operation/progress/ProgressProcessor.scala | 2 +-
4 files changed, 12 insertions(+), 9 deletions(-)
diff --git a/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java b/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java
index 7703f8b25..31e49f748 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java
+++ b/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java
@@ -311,7 +311,7 @@ public class EntranceRestfulApi implements EntranceRestfulRemote {
.reduce((x, y) -> x + y);
float corePercent = 0.0f;
float memoryPercent = 0.0f;
- if (cores.isPresent()) {
+ if (cores.isPresent() && memory.isPresent()) {
corePercent =
cores.get().floatValue()
/ EntranceConfiguration.YARN_QUEUE_CORES_MAX()
diff --git a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/execute/DefaultEntranceExecutor.scala b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/execute/DefaultEntranceExecutor.scala
index c6165523c..1a1fed310 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/execute/DefaultEntranceExecutor.scala
+++ b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/execute/DefaultEntranceExecutor.scala
@@ -90,8 +90,10 @@ class DefaultEntranceExecutor(id: Long, mark: MarkReq, entranceExecutorManager:
if (runningIndex >= 0 && runningIndex <= jobGroupSize -1) {
val totalProgress = 1.0 * (runningIndex + progressInfoEvent.progress) / jobGroupSize
//Update progress info
- progressInfoEvent.progressInfo.foreach(progressInfo =>
- subJobInfo.getProgressInfoMap.put(progressInfo.id, progressInfo))
+ if (null != progressInfoEvent.progressInfo) {
+ progressInfoEvent.progressInfo.foreach(progressInfo =>
+ subJobInfo.getProgressInfoMap.put(progressInfo.id, progressInfo))
+ }
entranceJob.getProgressListener.foreach(_.onProgressUpdate(entranceJob, totalProgress.toFloat,
entranceJob.getProgressInfo))
} else {
@@ -108,10 +110,11 @@ class DefaultEntranceExecutor(id: Long, mark: MarkReq, entranceExecutorManager:
}
val metricsMap = entranceJob.getJobRequest.getMetrics
val resourceMap = metricsMap.get(TaskConstant.ENTRANCEJOB_YARNRESOURCE)
+ val ecResourceMap = if (progressInfoEvent.resourceMap == null) new util.HashMap[String, ResourceWithStatus] else progressInfoEvent.resourceMap
if(resourceMap != null) {
- resourceMap.asInstanceOf[util.HashMap[String, ResourceWithStatus]].putAll(progressInfoEvent.resourceMap)
+ resourceMap.asInstanceOf[util.HashMap[String, ResourceWithStatus]].putAll(ecResourceMap)
} else {
- metricsMap.put(TaskConstant.ENTRANCEJOB_YARNRESOURCE, progressInfoEvent.resourceMap)
+ metricsMap.put(TaskConstant.ENTRANCEJOB_YARNRESOURCE, ecResourceMap)
}
// update engine info
// todo
@@ -123,9 +126,9 @@ class DefaultEntranceExecutor(id: Long, mark: MarkReq, entranceExecutorManager:
metricsMap.put(TaskConstant.ENTRANCEJOB_ENGINECONN_MAP, engineInstanceMap)
}
val infoMap = progressInfoEvent.infoMap
- if (infoMap.containsKey(TaskConstant.ENGINE_INSTANCE)) {
+ if (null != infoMap && infoMap.containsKey(TaskConstant.ENGINE_INSTANCE)) {
val instance = infoMap.get(TaskConstant.ENGINE_INSTANCE).asInstanceOf[String]
- val engineExtraInfoMap = engineInstanceMap.getOrDefault(instance, new util.HashMap[String, Object]).asInstanceOf[util.HashMap[String, Object]]
+ val engineExtraInfoMap = engineInstanceMap.getOrDefault(instance, new util.HashMap[String, Object]).asInstanceOf[util.HashMap[String, Object]]
engineInstanceMap.put(instance, engineExtraInfoMap)
engineExtraInfoMap.putAll(infoMap)
} else {
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
index f812976e9..0f492c138 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
@@ -55,7 +55,7 @@ class DefaultProgressOperation(orchestratorSession: OrchestratorSession) extends
/**
* To deal with the progress event from engineConn
- * @param taskRunningInfoEvent progress event
+ * @param oriTaskRunningInfoEvent progress event
*/
override def onProgressOn(oriTaskRunningInfoEvent: TaskRunningInfoEvent): Unit = {
var taskRunningInfoEvent: TaskRunningInfoEvent = null
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressProcessor.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressProcessor.scala
index 807c4f1ad..6a1fbd126 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressProcessor.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressProcessor.scala
@@ -41,7 +41,7 @@ class ProgressProcessor(rootExecTaskId: String,
}
def onProgress(progress: Float, progressInfo: Array[JobProgressInfo], resourceMap: util.HashMap[String, ResourceWithStatus], infoMap: util.HashMap[String, Object]): Unit = {
- val progressInfoWithResourceEvent = ProgressInfoEvent(orchestration, progress, progressInfo, resourceMap, infoMap: util.HashMap[String, Object])
+ val progressInfoWithResourceEvent = ProgressInfoEvent(orchestration, progress, progressInfo, resourceMap, infoMap)
listeners.foreach(_(progressInfoWithResourceEvent))
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[incubator-linkis] 02/09: linkis-computation-orchestrator - fix
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch dev-1.1.3
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
commit f49cd0759c83c21c7636815f478cde6b9ac85a92
Author: alexkun <xu...@qq.com>
AuthorDate: Thu May 12 20:12:23 2022 +0800
linkis-computation-orchestrator - fix
---
.../operation/progress/DefaultProgressOperation.scala | 9 +++++----
.../computation/operation/progress/ProgressProcessor.scala | 2 +-
2 files changed, 6 insertions(+), 5 deletions(-)
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
index 88f5fb90a..4f6819cc7 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/DefaultProgressOperation.scala
@@ -61,11 +61,12 @@ class DefaultProgressOperation(orchestratorSession: OrchestratorSession) extends
var taskRunningInfoEvent: TaskRunningInfoEvent = null
val execTaskInfo = generateExecTaskInfo(oriTaskRunningInfoEvent.execTask)
if (null != taskRunningInfoEvent.infoMap) {
+ taskRunningInfoEvent.infoMap.putAll(execTaskInfo)
taskRunningInfoEvent = TaskRunningInfoEvent(oriTaskRunningInfoEvent.execTask, oriTaskRunningInfoEvent.progress,
- oriTaskRunningInfoEvent.progressInfo, oriTaskRunningInfoEvent.resourceMap, ta)
+ oriTaskRunningInfoEvent.progressInfo, oriTaskRunningInfoEvent.resourceMap, taskRunningInfoEvent.infoMap)
} else {
taskRunningInfoEvent = TaskRunningInfoEvent(oriTaskRunningInfoEvent.execTask, oriTaskRunningInfoEvent.progress,
- oriTaskRunningInfoEvent.progressInfo, oriTaskRunningInfoEvent.resourceMap, generateExecTaskInfo(oriTaskRunningInfoEvent.execTask))
+ oriTaskRunningInfoEvent.progressInfo, oriTaskRunningInfoEvent.resourceMap, execTaskInfo)
}
val execTask = taskRunningInfoEvent.execTask
Option(execTask).foreach(task => {
@@ -100,7 +101,7 @@ class DefaultProgressOperation(orchestratorSession: OrchestratorSession) extends
}
Option(execTaskToProgressProcessor.get(execTask.getPhysicalContext.getRootTask.getId)).foreach( progress => {
- progress.onProgress(event.progress, event.progressInfo, event.resourceMap, event.resourceMap, event.infoMap)
+ progress.onProgress(event.progress, event.progressInfo, event.resourceMap, event.infoMap)
})
})
@@ -112,7 +113,7 @@ class DefaultProgressOperation(orchestratorSession: OrchestratorSession) extends
logger.warn("TaskRunningInfoEvent got null execTask.")
return map
}
- map.put(TaskConstants.EXEC_ID, task.getId())
+ map.put(TaskConstants.EXEC_ID, task.getId)
map.put(TaskConstants.ID_INFO, task.getIDInfo())
map.put(TaskConstants.TASK_CLASSNAME, task.getClass.getSimpleName)
map.put(TaskConstants.TASK_NAME, task.getName)
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressProcessor.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressProcessor.scala
index 680044651..807c4f1ad 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressProcessor.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/operation/progress/ProgressProcessor.scala
@@ -41,7 +41,7 @@ class ProgressProcessor(rootExecTaskId: String,
}
def onProgress(progress: Float, progressInfo: Array[JobProgressInfo], resourceMap: util.HashMap[String, ResourceWithStatus], infoMap: util.HashMap[String, Object]): Unit = {
- val progressInfoWithResourceEvent = ProgressInfoEvent(orchestration, progress, progressInfo, resourceMap)
+ val progressInfoWithResourceEvent = ProgressInfoEvent(orchestration, progress, progressInfo, resourceMap, infoMap: util.HashMap[String, Object])
listeners.foreach(_(progressInfoWithResourceEvent))
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[incubator-linkis] 04/09: 1. linkis-entrance - fix method progressWithResource
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch dev-1.1.3
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
commit f0780ad11d43862cdec8f98cd0d0cc5ea0484bbf
Author: alexkun <xu...@qq.com>
AuthorDate: Mon May 30 23:36:40 2022 +0800
1. linkis-entrance - fix method progressWithResource
---
.../apache/linkis/entrance/restful/EntranceRestfulApi.java | 12 +++++++-----
1 file changed, 7 insertions(+), 5 deletions(-)
diff --git a/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java b/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java
index 1e4862755..11d8958f3 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java
+++ b/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java
@@ -294,11 +294,13 @@ public class EntranceRestfulApi implements EntranceRestfulRemote {
(HashMap<String, ResourceWithStatus>)
metrics.get(TaskConstant.ENTRANCEJOB_YARNRESOURCE);
ArrayList<YarnResourceWithStatusVo> resoureList = new ArrayList<>(12);
- resourceMap.forEach(
- (applicationId, resource) -> {
- resoureList.add(
- new YarnResourceWithStatusVo(applicationId, resource));
- });
+ if (null != resourceMap && !resourceMap.isEmpty()) {
+ resourceMap.forEach(
+ (applicationId, resource) -> {
+ resoureList.add(
+ new YarnResourceWithStatusVo(applicationId, resource));
+ });
+ }
metricsVo.put(TaskConstant.ENTRANCEJOB_YARNRESOURCE, resoureList);
Optional<Integer> cores =
resourceMap.values().stream()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[incubator-linkis] 07/09: format code
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch dev-1.1.3
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
commit 185f027f1237b86434d855164d0af11ac16a6d81
Author: alexkun <xu...@qq.com>
AuthorDate: Tue Jun 7 17:38:05 2022 +0800
format code
---
.../apache/linkis/entrance/restful/EntranceRestfulApi.java | 13 +++++++------
1 file changed, 7 insertions(+), 6 deletions(-)
diff --git a/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java b/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java
index 62f425b16..7703f8b25 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java
+++ b/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceRestfulApi.java
@@ -314,15 +314,16 @@ public class EntranceRestfulApi implements EntranceRestfulRemote {
if (cores.isPresent()) {
corePercent =
cores.get().floatValue()
- / EntranceConfiguration.YARN_QUEUE_CORES_MAX().getValue();
+ / EntranceConfiguration.YARN_QUEUE_CORES_MAX()
+ .getValue();
memoryPercent =
memory.get().floatValue()
/ (EntranceConfiguration.YARN_QUEUE_MEMORY_MAX()
- .getValue()
- .longValue()
- * 1024
- * 1024
- * 1024);
+ .getValue()
+ .longValue()
+ * 1024
+ * 1024
+ * 1024);
}
String coreRGB = RGBUtils.getRGB(corePercent);
String memoryRGB = RGBUtils.getRGB(memoryPercent);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org