You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@linkis.apache.org by al...@apache.org on 2022/04/06 02:36:51 UTC
[incubator-linkis] branch dev-1.1.1 updated (6ba0958bd -> a337bc4f3)
This is an automated email from the ASF dual-hosted git repository.
alexkun pushed a change to branch dev-1.1.1
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
from 6ba0958bd No suitable ExternalResourceProvider found for cluster
new e3b8d34ba The group cache in EntranceFactory is replaced with guava cache #1901
new 2a83867f8 update am reuse default for 2
new a337bc4f3 update consumer idle default value
The 3 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:
.../scheduler/conf/SchedulerConfiguration.scala | 4 +-
.../parallelqueue/ParallelConsumerManager.scala | 8 ++--
.../entrance/conf/EntranceConfiguration.scala | 6 ++-
.../entrance/scheduler/EntranceGroupFactory.scala | 48 +++++++++++-----------
.../linkis/manager/am/conf/AMConfiguration.scala | 2 +-
5 files changed, 38 insertions(+), 30 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[incubator-linkis] 03/03: update consumer idle default value
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexkun pushed a commit to branch dev-1.1.1
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
commit a337bc4f3700995f0eccf78de4550060db5a0887
Author: peacewong <wp...@gmail.com>
AuthorDate: Mon Apr 4 10:48:03 2022 +0800
update consumer idle default value
---
.../org/apache/linkis/scheduler/conf/SchedulerConfiguration.scala | 4 ++--
.../scheduler/queue/parallelqueue/ParallelConsumerManager.scala | 8 +++++---
2 files changed, 7 insertions(+), 5 deletions(-)
diff --git a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/conf/SchedulerConfiguration.scala b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/conf/SchedulerConfiguration.scala
index c14bf0f3c..05dba4430 100644
--- a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/conf/SchedulerConfiguration.scala
+++ b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/conf/SchedulerConfiguration.scala
@@ -23,9 +23,9 @@ object SchedulerConfiguration {
val FIFO_CONSUMER_AUTO_CLEAR_ENABLED = CommonVars("wds.linkis.fifo.consumer.auto.clear.enabled", true)
- val FIFO_CONSUMER_MAX_IDLE_TIME = CommonVars("wds.linkis.fifo.consumer.max.idle.time", new TimeType("2h")).getValue.toLong
+ val FIFO_CONSUMER_MAX_IDLE_TIME = CommonVars("wds.linkis.fifo.consumer.max.idle.time", new TimeType("1h")).getValue.toLong
- val FIFO_CONSUMER_IDLE_SCAN_INTERVAL = CommonVars("wds.linkis.fifo.consumer.idle.scan.interval", new TimeType("6h"))
+ val FIFO_CONSUMER_IDLE_SCAN_INTERVAL = CommonVars("wds.linkis.fifo.consumer.idle.scan.interval", new TimeType("2h"))
val FIFO_CONSUMER_IDLE_SCAN_INIT_TIME = CommonVars("wds.linkis.fifo.consumer.idle.scan.init.time", new TimeType("1s"))
diff --git a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/parallelqueue/ParallelConsumerManager.scala b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/parallelqueue/ParallelConsumerManager.scala
index 889599d9e..c8fb5360f 100644
--- a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/parallelqueue/ParallelConsumerManager.scala
+++ b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/parallelqueue/ParallelConsumerManager.scala
@@ -51,9 +51,11 @@ class ParallelConsumerManager(maxParallelismUsers: Int, schedulerName: String) e
override def run(): Unit = CONSUMER_LOCK.synchronized {
info("Start to Clean up idle consumers ")
val nowTime = System.currentTimeMillis()
- consumerGroupMap.values.filter(_.isIdle)
- .filter(consumer => nowTime - consumer.getLastTime > SchedulerConfiguration.FIFO_CONSUMER_MAX_IDLE_TIME)
- .foreach(consumer => destroyConsumer(consumer.getGroup.getGroupName))
+ Utils.tryAndWarn {
+ consumerGroupMap.values.filter(_.isIdle)
+ .filter(consumer => nowTime - consumer.getLastTime > SchedulerConfiguration.FIFO_CONSUMER_MAX_IDLE_TIME)
+ .foreach(consumer => destroyConsumer(consumer.getGroup.getGroupName))
+ }
info(s"Finished to clean up idle consumers for $schedulerName, cost ${System.currentTimeMillis() - nowTime} ms.")
}
},
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[incubator-linkis] 02/03: update am reuse default for 2
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexkun pushed a commit to branch dev-1.1.1
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
commit 2a83867f82342b805a9cd3b3fc6c1e6cbe81b25d
Author: peacewong <wp...@gmail.com>
AuthorDate: Mon Apr 4 10:47:18 2022 +0800
update am reuse default for 2
---
.../main/scala/org/apache/linkis/manager/am/conf/AMConfiguration.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/conf/AMConfiguration.scala b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/conf/AMConfiguration.scala
index 891961297..798013e0e 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/conf/AMConfiguration.scala
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/conf/AMConfiguration.scala
@@ -34,7 +34,7 @@ object AMConfiguration {
val ENGINE_REUSE_MAX_TIME = CommonVars("wds.linkis.manager.am.engine.reuse.max.time", new TimeType("5m"))
- val ENGINE_REUSE_COUNT_LIMIT = CommonVars("wds.linkis.manager.am.engine.reuse.count.limit", 10)
+ val ENGINE_REUSE_COUNT_LIMIT = CommonVars("wds.linkis.manager.am.engine.reuse.count.limit", 2)
val NODE_STATUS_HEARTBEAT_TIME = CommonVars("wds.linkis.manager.am.node.heartbeat", new TimeType("3m"))
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[incubator-linkis] 01/03: The group cache in EntranceFactory is replaced with guava cache #1901
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexkun pushed a commit to branch dev-1.1.1
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
commit e3b8d34ba0b82727db0261cc4a15a44f5c0c8517
Author: peacewong <wp...@gmail.com>
AuthorDate: Mon Apr 4 10:46:41 2022 +0800
The group cache in EntranceFactory is replaced with guava cache #1901
---
.../entrance/conf/EntranceConfiguration.scala | 6 ++-
.../entrance/scheduler/EntranceGroupFactory.scala | 48 +++++++++++-----------
2 files changed, 30 insertions(+), 24 deletions(-)
diff --git a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/conf/EntranceConfiguration.scala b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/conf/EntranceConfiguration.scala
index df4ac7d01..916bcc8a3 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/conf/EntranceConfiguration.scala
+++ b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/conf/EntranceConfiguration.scala
@@ -83,7 +83,7 @@ object EntranceConfiguration {
* wds.linkis.dwc.instance is a parameter used to control the number of engines each user starts.
*wds.linkis.instance 是用来进行控制每个用户启动engine数量的参数
*/
- val WDS_LINKIS_INSTANCE = CommonVars("wds.linkis.rm.instance", 3)
+ val WDS_LINKIS_INSTANCE = CommonVars("wds.linkis.rm.instance", 10)
val LOG_EXCLUDE_ALL = CommonVars("wds.linkis.log.exclude.all", "com.netflix")
@@ -172,4 +172,8 @@ object EntranceConfiguration {
val CLI_HEARTBEAT_THRESHOLD_SECONDS = CommonVars[Long] ("linkis.entrance.cli.heartbeat.threshold.sec", 30l).getValue
val LOG_PUSH_INTERVAL_TIME = CommonVars("wds.linkis.entrance.log.push.interval.time", 5 * 60 * 1000)
+
+ val GRORUP_CACHE_MAX = CommonVars("wds.linkis.consumer.group.cache.capacity", 5000)
+
+ val GRORUP_CACHE_EXPITE_TIME = CommonVars("wds.linkis.consumer.group.expire.time.hour", 50)
}
diff --git a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/scheduler/EntranceGroupFactory.scala b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/scheduler/EntranceGroupFactory.scala
index 6e925711f..1db867b3f 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/scheduler/EntranceGroupFactory.scala
+++ b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/scheduler/EntranceGroupFactory.scala
@@ -17,33 +17,36 @@
package org.apache.linkis.entrance.scheduler
-import java.util
-
+import com.google.common.cache.{Cache, CacheBuilder}
+import org.apache.commons.lang.StringUtils
import org.apache.linkis.common.conf.{CommonVars, Configuration}
import org.apache.linkis.common.utils.{Logging, Utils}
import org.apache.linkis.entrance.conf.EntranceConfiguration
import org.apache.linkis.entrance.exception.{EntranceErrorCode, EntranceErrorException}
import org.apache.linkis.entrance.execute.EntranceJob
-import org.apache.linkis.governance.common.protocol.conf.{RequestQueryEngineConfig, RequestQueryEngineConfigWithGlobalConfig, ResponseQueryConfig}
+import org.apache.linkis.governance.common.protocol.conf.{RequestQueryEngineConfigWithGlobalConfig, ResponseQueryConfig}
import org.apache.linkis.manager.label.entity.Label
import org.apache.linkis.manager.label.entity.engine.{ConcurrentEngineConnLabel, EngineTypeLabel, UserCreatorLabel}
+import org.apache.linkis.manager.label.utils.LabelUtil
import org.apache.linkis.protocol.constants.TaskConstant
import org.apache.linkis.protocol.utils.TaskUtils
import org.apache.linkis.rpc.Sender
import org.apache.linkis.scheduler.queue.parallelqueue.ParallelGroup
import org.apache.linkis.scheduler.queue.{Group, GroupFactory, SchedulerEvent}
-import org.apache.linkis.server.JMap
-import org.apache.commons.lang.StringUtils
+import java.util
+import java.util.concurrent.TimeUnit
import scala.collection.JavaConversions._
class EntranceGroupFactory extends GroupFactory with Logging {
- private val groupNameToGroups = new JMap[String, Group]
-// private val labelBuilderFactory = LabelBuilderFactoryContext.getLabelBuilderFactory
+
+ private val groupNameToGroups: Cache[String, Group] = CacheBuilder.newBuilder().expireAfterAccess(EntranceConfiguration.GRORUP_CACHE_EXPITE_TIME.getValue, TimeUnit.MINUTES)
+ .maximumSize(EntranceConfiguration.GRORUP_CACHE_MAX.getValue).build()
private val GROUP_MAX_CAPACITY = CommonVars("wds.linkis.entrance.max.capacity", 2000)
+
private val GROUP_INIT_CAPACITY = CommonVars("wds.linkis.entrance.init.capacity", 100)
@@ -53,11 +56,12 @@ class EntranceGroupFactory extends GroupFactory with Logging {
(job.getJobRequest.getLabels, job.getJobRequest.getParams.asInstanceOf[util.Map[String, Any]])
}
val groupName = EntranceGroupFactory.getGroupNameByLabels(labels, params)
- if (!groupNameToGroups.containsKey(groupName)) synchronized {
+ val cacheGroup = groupNameToGroups.getIfPresent(groupName)
+ if (null == cacheGroup) synchronized {
val maxAskExecutorTimes = EntranceConfiguration.MAX_ASK_EXECUTOR_TIME.getValue.toLong
if (groupName.startsWith(EntranceGroupFactory.CONCURRENT)) {
- if (null == groupNameToGroups.get(groupName)) synchronized {
- if (null == groupNameToGroups.get(groupName)) {
+ if (null == groupNameToGroups.getIfPresent(groupName)) synchronized {
+ if (null == groupNameToGroups.getIfPresent(groupName)) {
val group = new ParallelGroup(groupName, 100, EntranceConfiguration.CONCURRENT_FACTORY_MAX_CAPACITY.getValue)
group.setMaxRunningJobs(EntranceConfiguration.CONCURRENT_MAX_RUNNING_JOBS.getValue)
group.setMaxAskExecutorTimes(EntranceConfiguration.CONCURRENT_EXECUTOR_TIME.getValue)
@@ -67,36 +71,34 @@ class EntranceGroupFactory extends GroupFactory with Logging {
}
}
val sender: Sender = Sender.getSender(Configuration.CLOUD_CONSOLE_CONFIGURATION_SPRING_APPLICATION_NAME.getValue)
- var userCreatorLabel: UserCreatorLabel = null
- var engineTypeLabel: EngineTypeLabel = null
- labels.foreach {
- case label: UserCreatorLabel => userCreatorLabel = label
- case label: EngineTypeLabel => engineTypeLabel = label
- case _ =>
- }
- info(s"Getting user configurations for $groupName(正在为 $groupName 获取参数) userCreatorLabel: ${userCreatorLabel.getStringValue}, engineTypeLabel:${engineTypeLabel.getStringValue}.")
+ val userCreatorLabel: UserCreatorLabel = LabelUtil.getUserCreatorLabel(labels)
+ val engineTypeLabel: EngineTypeLabel = LabelUtil.getEngineTypeLabel(labels)
+ logger.info(s"Getting user configurations for $groupName userCreatorLabel: ${userCreatorLabel.getStringValue}, engineTypeLabel:${engineTypeLabel.getStringValue}.")
val keyAndValue = Utils.tryAndWarnMsg {
sender.ask(RequestQueryEngineConfigWithGlobalConfig(userCreatorLabel, engineTypeLabel)).asInstanceOf[ResponseQueryConfig].getKeyAndValue
}("Get user configurations from configuration server failed! Next use the default value to continue.")
val maxRunningJobs = EntranceConfiguration.WDS_LINKIS_INSTANCE.getValue(keyAndValue)
val initCapacity = GROUP_INIT_CAPACITY.getValue(keyAndValue)
val maxCapacity = GROUP_MAX_CAPACITY.getValue(keyAndValue)
- info(s"Got user configurations: groupName=$groupName, maxRunningJobs=$maxRunningJobs, initCapacity=$initCapacity, maxCapacity=$maxCapacity.")
+ logger.info(s"Got user configurations: groupName=$groupName, maxRunningJobs=$maxRunningJobs, initCapacity=$initCapacity, maxCapacity=$maxCapacity.")
val group = new ParallelGroup(groupName, initCapacity, maxCapacity)
group.setMaxRunningJobs(maxRunningJobs)
group.setMaxAskExecutorTimes(maxAskExecutorTimes)
- if (!groupNameToGroups.containsKey(groupName)) groupNameToGroups.put(groupName, group)
+ groupNameToGroups.put(groupName, group)
}
- groupNameToGroups.get(groupName)
+ groupNameToGroups.getIfPresent(groupName)
}
override def getGroup(groupName: String): Group = {
- val group = groupNameToGroups.get(groupName)
- if(group == null){
+ val group = groupNameToGroups.getIfPresent(groupName)
+ if (group == null) {
throw new EntranceErrorException(EntranceErrorCode.GROUP_NOT_FOUND.getErrCode, s"group not found: ${groupName}")
}
group
}
+
+
+
}
object EntranceGroupFactory {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org