From adb057987ffb6cf212c5f4346d72fbfa248de1c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E4=B8=87=E4=B9=89?= Date: Tue, 8 Sep 2026 16:39:44 +0800 Subject: [PATCH] [BugFix] Update parallelgroup cache when entrance instance goes offline (#5407) --- .../restful/EntranceLabelRestfulApi.java | 32 ++++++++- .../EntranceOrchestrationFactory.scala | 23 +++++++ ...tranceUserParallelOrchestratorPlugin.scala | 6 ++ .../scheduler/EntranceGroupFactory.scala | 66 +++++++++++++++++++ 4 files changed, 126 insertions(+), 1 deletion(-) diff --git a/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceLabelRestfulApi.java b/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceLabelRestfulApi.java index 8a0d0393d55..68a1132b6d8 100644 --- a/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceLabelRestfulApi.java +++ b/linkis-computation-governance/linkis-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceLabelRestfulApi.java @@ -21,6 +21,7 @@ import org.apache.linkis.common.ServiceInstance; import org.apache.linkis.common.conf.Configuration; import org.apache.linkis.entrance.EntranceServer; +import org.apache.linkis.entrance.orchestrator.EntranceOrchestrationFactory; import org.apache.linkis.entrance.scheduler.EntranceSchedulerContext; import org.apache.linkis.instance.label.client.InstanceLabelClient; import org.apache.linkis.manager.label.constant.LabelKeyConstant; @@ -106,7 +107,17 @@ public Message updateRouteLabel(HttpServletRequest req) { SchedulerContext schedulerContext = entranceServer.getEntranceContext().getOrCreateScheduler().getSchedulerContext(); if (schedulerContext instanceof EntranceSchedulerContext) { - ((EntranceSchedulerContext) schedulerContext).setOfflineFlag(true); + EntranceSchedulerContext entranceSchedulerContext = (EntranceSchedulerContext) schedulerContext; + entranceSchedulerContext.setOfflineFlag(true); + if (entranceSchedulerContext.getOrCreateGroupFactory() + instanceof org.apache.linkis.entrance.scheduler.EntranceGroupFactory) { + ((org.apache.linkis.entrance.scheduler.EntranceGroupFactory) + entranceSchedulerContext.getOrCreateGroupFactory()) + .refreshAllGroups(); + invalidateUserParallelCache(); + logger.info( + "Refreshed scheduler group cache and user parallel plugin cache after mark offline."); + } } entranceServer.updateAllNotExecutionTaskInstances(true); logger.info("Finished to update all not execution task instances to empty string"); @@ -128,10 +139,29 @@ public Message backOnline(HttpServletRequest req) { synchronized (offlineFlag) { // NOSONAR offlineFlag = false; } + if (entranceServer.getEntranceContext().getOrCreateScheduler().getSchedulerContext() + instanceof EntranceSchedulerContext) { + EntranceSchedulerContext entranceSchedulerContext = + (EntranceSchedulerContext) + entranceServer.getEntranceContext().getOrCreateScheduler().getSchedulerContext(); + if (entranceSchedulerContext.getOrCreateGroupFactory() + instanceof org.apache.linkis.entrance.scheduler.EntranceGroupFactory) { + ((org.apache.linkis.entrance.scheduler.EntranceGroupFactory) + entranceSchedulerContext.getOrCreateGroupFactory()) + .refreshAllGroups(); + invalidateUserParallelCache(); + logger.info( + "Refreshed scheduler group cache and user parallel plugin cache after back online."); + } + } logger.info("Finished to backonline"); return Message.ok(); } + private void invalidateUserParallelCache() { + new EntranceOrchestrationFactory().invalidateUserParallelPluginCache(); + } + @ApiOperation(value = "isOnline", notes = "entrance isOnline", response = Message.class) @RequestMapping(path = "/isOnline", method = RequestMethod.GET) public Message isOnline(HttpServletRequest req) { 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 1f7f3158517..5186ad50372 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 @@ -52,4 +52,27 @@ object EntranceOrchestrationFactory { def getOrchestrationSession(): OrchestratorSession = orchestratorSession + /** + * Invalidate the cached user max running jobs so that they are recalculated with the latest + * number of running entrance instances (e.g. after entrances go offline/back online). + */ + def invalidateUserParallelPluginCache(): Unit = + orchestratorSession.orchestrator.getOrchestratorContext match { + case ctx: AbstractOrchestratorContext => + ctx.getOrchestratorPlugins.foreach { + case plugin: EntranceUserParallelOrchestratorPlugin => plugin.invalidateCache() + case _ => + } + case _ => + } + +} + +/** + * Java-visible companion of the `EntranceOrchestrationFactory` object. Scala 2.11 does not emit + * static forwarders for objects, so Java callers route through an instance method here. + */ +class EntranceOrchestrationFactory { + def invalidateUserParallelPluginCache(): Unit = + EntranceOrchestrationFactory.invalidateUserParallelPluginCache() } diff --git a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/orchestrator/plugin/EntranceUserParallelOrchestratorPlugin.scala b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/orchestrator/plugin/EntranceUserParallelOrchestratorPlugin.scala index 1a2056be250..c8986c73338 100644 --- a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/orchestrator/plugin/EntranceUserParallelOrchestratorPlugin.scala +++ b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/orchestrator/plugin/EntranceUserParallelOrchestratorPlugin.scala @@ -97,6 +97,12 @@ class EntranceUserParallelOrchestratorPlugin extends UserParallelOrchestratorPlu configCache.get(EntranceUtils.getUserCreatorEcTypeKey(userCreatorLabel, engineTypeLabel)) } + /** + * Invalidate the cached user max running jobs, so that they are recalculated with the latest + * number of running entrance instances (e.g. after entrances go offline/back online). + */ + def invalidateCache(): Unit = this.configCache.invalidateAll() + override def isReady: Boolean = true override def start(): Unit = {} 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 cb1b610e2b5..51f6b966e53 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 @@ -28,6 +28,8 @@ import org.apache.linkis.governance.common.protocol.conf.{ RequestQueryEngineConfigWithGlobalConfig, ResponseQueryConfig } +import org.apache.linkis.manager.label.builder.factory.LabelBuilderFactoryContext +import org.apache.linkis.manager.label.constant.LabelKeyConstant import org.apache.linkis.manager.label.entity.Label import org.apache.linkis.manager.label.entity.engine.{EngineTypeLabel, UserCreatorLabel} import org.apache.linkis.manager.label.utils.LabelUtil @@ -44,14 +46,38 @@ import java.util.regex.Pattern import com.google.common.cache.{Cache, CacheBuilder} +import scala.collection.mutable +import scala.collection.JavaConverters._ + class EntranceGroupFactory extends GroupFactory with Logging { + case class GroupMaxRunningJobsKey(creator: String, user: String, engineType: String) + + private val labelFactory = LabelBuilderFactoryContext.getLabelBuilderFactory + private val groupNameToGroups: Cache[String, Group] = CacheBuilder .newBuilder() .expireAfterAccess(EntranceConfiguration.GROUP_CACHE_EXPIRE_TIME.getValue, TimeUnit.MINUTES) .maximumSize(EntranceConfiguration.GROUP_CACHE_MAX.getValue) .build() + /** creator_user_engineType -> GroupMaxRunningJobsKey, used to recalculate maxRunningJobs */ + private val groupNameToKey: mutable.Map[String, GroupMaxRunningJobsKey] = mutable.Map() + + private def getGroupMaxRunningJobsKey(labels: java.util.List[Label[_]]) + : GroupMaxRunningJobsKey = { + val userCreatorLabel: UserCreatorLabel = LabelUtil.getUserCreatorLabel(labels) + val engineTypeLabel: EngineTypeLabel = LabelUtil.getEngineTypeLabel(labels) + if (null == userCreatorLabel || null == engineTypeLabel) { + throw new EntranceErrorException(LABEL_NOT_NULL.getErrorCode, LABEL_NOT_NULL.getErrorDesc) + } + GroupMaxRunningJobsKey( + userCreatorLabel.getCreator, + userCreatorLabel.getUser, + engineTypeLabel.getEngineType + ) + } + private val GROUP_MAX_CAPACITY = CommonVars("wds.linkis.entrance.max.capacity", 1000) private val SPECIFIED_USERNAME_REGEX = @@ -120,6 +146,7 @@ class EntranceGroupFactory extends GroupFactory with Logging { group.setMaxRunningJobs(maxRunningJobs) group.setMaxAskExecutorTimes(maxAskExecutorTimes) groupNameToGroups.put(groupName, group) + groupNameToKey.put(groupName, getGroupMaxRunningJobsKey(labels)) group } else { @@ -127,6 +154,45 @@ class EntranceGroupFactory extends GroupFactory with Logging { } } + /** + * Recalculate the maxRunningJobs of a cached group with the latest user configuration and the + * latest number of running entrance instances, and set it to the group. + * @param key + * @return + */ + private def refreshMaxRunningJobs(key: GroupMaxRunningJobsKey): Unit = { + val groupName = s"${key.creator}_${key.user}_${key.engineType}" + val group = groupNameToGroups.getIfPresent(groupName) + if (null == group) return + val labels = labelFactory.createLabel[UserCreatorLabel](LabelKeyConstant.USER_CREATOR_TYPE_KEY) + labels.setCreator(key.creator) + labels.setUser(key.user) + val ecTypeLabel = labelFactory.createLabel[EngineTypeLabel](LabelKeyConstant.ENGINE_TYPE_KEY) + ecTypeLabel.setEngineType(key.engineType) + val sender: Sender = + Sender.getSender(Configuration.CLOUD_CONSOLE_CONFIGURATION_SPRING_APPLICATION_NAME.getValue) + val keyAndValue = Utils.tryAndWarnMsg { + sender + .ask(RequestQueryEngineConfigWithGlobalConfig(labels, ecTypeLabel)) + .asInstanceOf[ResponseQueryConfig] + .getKeyAndValue + }( + "Get user configurations from configuration server failed! Next use the default value to continue." + ) + val maxRunningJobs = EntranceGroupFactory.getUserMaxRunningJobs(keyAndValue) + group.asInstanceOf[ParallelGroup].setMaxRunningJobs(maxRunningJobs) + logger.info(s"Refreshed maxRunningJobs of group $groupName to $maxRunningJobs.") + } + + /** + * Refresh maxRunningJobs for all cached groups, so that after entrances go offline/back online + * the concurrency is recalculated with the latest number of running entrance instances. + */ + def refreshAllGroups(): Unit = + groupNameToGroups.asMap().keySet().asScala.foreach { groupName => + groupNameToKey.get(groupName).foreach(refreshMaxRunningJobs) + } + override def getGroup(groupName: String): Group = { val group = groupNameToGroups.getIfPresent(groupName) if (group == null) {