You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2023/02/28 11:33:33 UTC
[incubator-streampark] branch dev updated: [Bug] After switch groups, the metrics dashboard info is not switched… (#2376)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
The following commit(s) were added to refs/heads/dev by this push:
new fb54c0512 [Bug] After switch groups, the metrics dashboard info is not switched… (#2376)
fb54c0512 is described below
commit fb54c051232954caba22e77be182409d38fd1b17
Author: benjobs <be...@apache.org>
AuthorDate: Tue Feb 28 19:33:27 2023 +0800
[Bug] After switch groups, the metrics dashboard info is not switched… (#2376)
* [Bug] After switch groups, the metrics dashboard info is not switched correctly. #2315
---
.../core/controller/ApplicationController.java | 6 --
.../console/core/service/ApplicationService.java | 2 -
.../core/service/impl/ApplicationServiceImpl.java | 38 +++++-------
.../console/core/task/FlinkK8sWatcherWrapper.java | 12 +++-
.../src/api/flink/app/app.ts | 1 -
.../flink/kubernetes/DefaultFlinkK8sWatcher.scala | 5 +-
.../flink/kubernetes/FlinkK8sWatchController.scala | 22 +++----
.../flink/kubernetes/FlinkK8sWatcher.scala | 2 +-
.../kubernetes/FlinkK8sWatcherLazyStartAop.scala | 4 +-
.../helper/KubernetesWatcherHelper.scala | 2 +-
.../flink/kubernetes/model/FlinkMetricCV.scala | 69 ++++++++++++----------
.../flink/kubernetes/model/TrackId.scala | 26 ++++----
.../kubernetes/watcher/FlinkJobStatusWatcher.scala | 12 ++--
.../kubernetes/watcher/FlinkMetricsWatcher.scala | 1 +
14 files changed, 99 insertions(+), 103 deletions(-)
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java
index 52702ccb1..54e08c749 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java
@@ -377,12 +377,6 @@ public class ApplicationController {
return RestResponse.success(uploadPath);
}
- @PostMapping("downlog")
- public RestResponse downlog(Long id) {
- applicationService.tailMvnDownloading(id);
- return RestResponse.success();
- }
-
@PostMapping("verifySchema")
public RestResponse verifySchema(String path) {
final URI uri = URI.create(path);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationService.java
index 555e3e488..2ce616ad4 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationService.java
@@ -71,8 +71,6 @@ public interface ApplicationService extends IService<Application> {
Map<String, Serializable> dashboard(Long teamId);
- void tailMvnDownloading(Long id);
-
String upload(MultipartFile file) throws Exception;
/** set the latest to Effective, it will really become the current effective */
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
index 06dbd9aff..3fbc45514 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
@@ -154,10 +154,6 @@ import static org.apache.streampark.console.core.utils.YarnQueueLabelExpression.
public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Application>
implements ApplicationService {
- private final Map<Long, Long> tailOutMap = new ConcurrentHashMap<>();
-
- private final Map<Long, Boolean> tailBeginning = new ConcurrentHashMap<>();
-
private static final int DEFAULT_HISTORY_RECORD_LIMIT = 25;
private static final int DEFAULT_HISTORY_POD_TMPL_RECORD_LIMIT = 5;
@@ -269,19 +265,22 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
}
// merge metrics from flink kubernetes cluster
- FlinkMetricCV k8sMetric = k8SFlinkTrackMonitor.getAccClusterMetrics();
- totalJmMemory += k8sMetric.totalJmMemory();
- totalTmMemory += k8sMetric.totalTmMemory();
- totalTm += k8sMetric.totalTm();
- totalSlot += k8sMetric.totalSlot();
- availableSlot += k8sMetric.availableSlot();
- runningJob += k8sMetric.runningJob();
- overview.setTotal(overview.getTotal() + k8sMetric.totalJob());
- overview.setRunning(overview.getRunning() + k8sMetric.runningJob());
- overview.setFinished(overview.getFinished() + k8sMetric.finishedJob());
- overview.setCanceled(overview.getCanceled() + k8sMetric.cancelledJob());
- overview.setFailed(overview.getFailed() + k8sMetric.failedJob());
+ FlinkMetricCV k8sMetric = k8SFlinkTrackMonitor.getAccGroupMetrics(teamId.toString());
+ if (k8sMetric != null) {
+ totalJmMemory += k8sMetric.totalJmMemory();
+ totalTmMemory += k8sMetric.totalTmMemory();
+ totalTm += k8sMetric.totalTm();
+ totalSlot += k8sMetric.totalSlot();
+ availableSlot += k8sMetric.availableSlot();
+ runningJob += k8sMetric.runningJob();
+ overview.setTotal(overview.getTotal() + k8sMetric.totalJob());
+ overview.setRunning(overview.getRunning() + k8sMetric.runningJob());
+ overview.setFinished(overview.getFinished() + k8sMetric.finishedJob());
+ overview.setCanceled(overview.getCanceled() + k8sMetric.cancelledJob());
+ overview.setFailed(overview.getFailed() + k8sMetric.failedJob());
+ }
+ // result json
Map<String, Serializable> map = new HashMap<>(8);
map.put("task", overview);
map.put("jmMemory", totalJmMemory);
@@ -294,13 +293,6 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
return map;
}
- @Override
- public void tailMvnDownloading(Long id) {
- this.tailOutMap.put(id, id);
- // the first time, will be read from the beginning of the buffer. Only once
- this.tailBeginning.put(id, true);
- }
-
@Override
public String upload(MultipartFile file) throws Exception {
File temp = WebUtils.getAppTempDir();
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sWatcherWrapper.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sWatcherWrapper.java
index ad45b73df..090c83695 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sWatcherWrapper.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sWatcherWrapper.java
@@ -123,10 +123,18 @@ public class FlinkK8sWatcherWrapper {
Enumeration.Value mode = FlinkK8sExecuteMode.of(app.getExecutionModeEnum());
if (FlinkK8sExecuteMode.APPLICATION().equals(mode)) {
return TrackId.onApplication(
- app.getK8sNamespace(), app.getClusterId(), app.getId(), app.getJobId());
+ app.getK8sNamespace(),
+ app.getClusterId(),
+ app.getId(),
+ app.getJobId(),
+ app.getTeamId().toString());
} else if (FlinkK8sExecuteMode.SESSION().equals(mode)) {
return TrackId.onSession(
- app.getK8sNamespace(), app.getClusterId(), app.getId(), app.getJobId());
+ app.getK8sNamespace(),
+ app.getClusterId(),
+ app.getId(),
+ app.getJobId(),
+ app.getTeamId().toString());
} else {
throw new IllegalArgumentException(
"Illegal K8sExecuteMode, mode=" + app.getExecutionMode());
diff --git a/streampark-console/streampark-console-webapp/src/api/flink/app/app.ts b/streampark-console/streampark-console-webapp/src/api/flink/app/app.ts
index 054b22f33..cfe65df8b 100644
--- a/streampark-console/streampark-console-webapp/src/api/flink/app/app.ts
+++ b/streampark-console/streampark-console-webapp/src/api/flink/app/app.ts
@@ -46,7 +46,6 @@ enum APP_API {
ROLLBACK = '/flink/app/rollback',
REVOKE = '/flink/app/revoke',
OPTION_LOG = '/flink/app/optionlog',
- DOWN_LOG = '/flink/app/downlog',
CHECK_JAR = '/flink/app/checkjar',
VERIFY_SCHEMA = '/flink/app/verifySchema',
CHECK_SAVEPOINT_PATH = '/flink/app/checkSavepointPath',
diff --git a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/DefaultFlinkK8sWatcher.scala b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/DefaultFlinkK8sWatcher.scala
index 3e617e968..195e98af5 100644
--- a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/DefaultFlinkK8sWatcher.scala
+++ b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/DefaultFlinkK8sWatcher.scala
@@ -81,15 +81,14 @@ class DefaultFlinkK8sWatcher(conf: FlinkTrackConfig = FlinkTrackConfig.defaultCo
override def getAllJobStatus: Map[CacheKey, JobStatusCV] = watchController.jobStatuses.asMap()
- override def getAccClusterMetrics: FlinkMetricCV = watchController.collectAccMetric()
+ override def getAccGroupMetrics(groupId: String): FlinkMetricCV = watchController.collectAccGroupMetric(groupId)
override def getClusterMetrics(clusterKey: ClusterKey): Option[FlinkMetricCV] = Option(watchController.flinkMetrics.get(clusterKey))
override def getAllWatchingIds: Set[TrackId] = watchController.getAllWatchingIds()
override def checkIsInRemoteCluster(trackId: TrackId): Boolean = {
- if (!trackId.isLegal) false;
- else {
+ if (!trackId.isLegal) false; else {
val nonLost = (state: FlinkJobState.Value) => state != FlinkJobState.LOST || state != FlinkJobState.SILENT
trackId.executeMode match {
case SESSION =>
diff --git a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/FlinkK8sWatchController.scala b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/FlinkK8sWatchController.scala
index 64f5e5054..7e78b293e 100644
--- a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/FlinkK8sWatchController.scala
+++ b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/FlinkK8sWatchController.scala
@@ -18,11 +18,8 @@
package org.apache.streampark.flink.kubernetes
import java.util.concurrent.TimeUnit
-
import scala.collection.JavaConversions._
-
import com.github.benmanes.caffeine.cache.{Cache, Caffeine}
-
import org.apache.streampark.common.util.Logger
import org.apache.streampark.flink.kubernetes.model._
@@ -63,8 +60,7 @@ class FlinkK8sWatchController extends Logger with AutoCloseable {
* determines whether the specified TrackId is in the trace
*/
def isInWatching(trackId: TrackId): Boolean = {
- if (!trackId.isLegal) false;
- else {
+ if (!trackId.isLegal) false; else {
trackIds.get(trackId) != null
}
}
@@ -87,18 +83,16 @@ class FlinkK8sWatchController extends Logger with AutoCloseable {
/**
* collect the aggregation of flink metrics that in tracking
*/
- def collectAccMetric(): FlinkMetricCV = {
+ def collectAccGroupMetric(groupId: String): FlinkMetricCV = {
// get cluster metrics that in tracking
+ val empty = FlinkMetricCV.empty(groupId)
getActiveWatchingIds() match {
- case k if k.isEmpty => FlinkMetricCV.empty
+ case k if k.isEmpty => empty
case k =>
- flinkMetrics.getAll(for (elem <- k) yield {
- ClusterKey.of(elem)
- }) match {
- case m if m.isEmpty => FlinkMetricCV.empty
- case m =>
- // aggregate metrics
- m.values.fold(FlinkMetricCV.empty)((x, y) => x + y)
+ flinkMetrics.getAll(for (elem <- k) yield ClusterKey.of(elem))
+ match {
+ case m if m.isEmpty => empty
+ case m => m.values.fold(empty)((x, y) => x + y)
}
}
}
diff --git a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/FlinkK8sWatcher.scala b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/FlinkK8sWatcher.scala
index a8e35b2d7..91994d6a6 100644
--- a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/FlinkK8sWatcher.scala
+++ b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/FlinkK8sWatcher.scala
@@ -104,7 +104,7 @@ trait FlinkK8sWatcher extends Logger with AutoCloseable {
/**
* get flink cluster metrics aggregation
*/
- def getAccClusterMetrics: FlinkMetricCV
+ def getAccGroupMetrics(groupId: String = null): FlinkMetricCV
/**
* get flink cluster metrics
diff --git a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/FlinkK8sWatcherLazyStartAop.scala b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/FlinkK8sWatcherLazyStartAop.scala
index a54ae480c..deac6dbe8 100644
--- a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/FlinkK8sWatcherLazyStartAop.scala
+++ b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/FlinkK8sWatcherLazyStartAop.scala
@@ -62,9 +62,9 @@ trait FlinkK8sWatcherLazyStartAop extends FlinkK8sWatcher {
super.getAllJobStatus
}
- abstract override def getAccClusterMetrics: FlinkMetricCV = {
+ abstract override def getAccGroupMetrics(groupId: String): FlinkMetricCV = {
// behavior of getting cluster metrics will not trigger a delayed start
- super.getAccClusterMetrics
+ super.getAccGroupMetrics(groupId)
}
abstract override def getClusterMetrics(clusterKey: ClusterKey): Option[FlinkMetricCV] = {
diff --git a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/helper/KubernetesWatcherHelper.scala b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/helper/KubernetesWatcherHelper.scala
index 4b89ae022..16ffdc761 100644
--- a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/helper/KubernetesWatcherHelper.scala
+++ b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/helper/KubernetesWatcherHelper.scala
@@ -47,7 +47,7 @@ object KubernetesWatcherHelper {
new Timer().scheduleAtFixedRate(
() =>
println(s"[flink-k8s][agg-metric]-${System.currentTimeMillis} => " +
- s"${k8sWatcher.getAccClusterMetrics}"),
+ s"${k8sWatcher.getAccGroupMetrics()}"),
0,
1500)
diff --git a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/model/FlinkMetricCV.scala b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/model/FlinkMetricCV.scala
index 419e5862c..73afe4d79 100644
--- a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/model/FlinkMetricCV.scala
+++ b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/model/FlinkMetricCV.scala
@@ -21,47 +21,54 @@ package org.apache.streampark.flink.kubernetes.model
* flink cluster metric info
*/
case class FlinkMetricCV(
- totalJmMemory: Integer = 0,
- totalTmMemory: Integer = 0,
- totalTm: Integer = 0,
- totalSlot: Integer = 0,
- availableSlot: Integer = 0,
- runningJob: Integer = 0,
- finishedJob: Integer = 0,
- cancelledJob: Integer = 0,
- failedJob: Integer = 0,
- pollAckTime: Long) {
+ groupId: String = null,
+ totalJmMemory: Integer = 0,
+ totalTmMemory: Integer = 0,
+ totalTm: Integer = 0,
+ totalSlot: Integer = 0,
+ availableSlot: Integer = 0,
+ runningJob: Integer = 0,
+ finishedJob: Integer = 0,
+ cancelledJob: Integer = 0,
+ failedJob: Integer = 0,
+ pollAckTime: Long) {
def +(another: FlinkMetricCV): FlinkMetricCV = {
- this.copy(
- totalJmMemory + another.totalJmMemory,
- totalTmMemory + another.totalTmMemory,
- totalTm + another.totalTm,
- totalSlot + another.totalSlot,
- availableSlot + another.availableSlot,
- runningJob + another.runningJob,
- finishedJob + another.finishedJob,
- cancelledJob + another.cancelledJob,
- failedJob + another.failedJob,
- pollAckTime = math.max(pollAckTime, another.pollAckTime))
+ if (another == null) this else {
+ if (this.groupId == null || this.groupId == another.groupId) {
+ this.copy(
+ groupId = this.groupId,
+ totalJmMemory + another.totalJmMemory,
+ totalTmMemory + another.totalTmMemory,
+ totalTm + another.totalTm,
+ totalSlot + another.totalSlot,
+ availableSlot + another.availableSlot,
+ runningJob + another.runningJob,
+ finishedJob + another.finishedJob,
+ cancelledJob + another.cancelledJob,
+ failedJob + another.failedJob,
+ pollAckTime = math.max(pollAckTime, another.pollAckTime))
+ } else this
+ }
}
def totalJob(): Integer = runningJob + finishedJob + cancelledJob + failedJob
def equalsPayload(another: FlinkMetricCV): Boolean = {
- totalJmMemory == another.totalTmMemory &&
- totalTmMemory == another.totalTmMemory &&
- totalTm == another.totalTm &&
- totalSlot == another.totalSlot &&
- availableSlot == another.availableSlot &&
- runningJob == another.runningJob &&
- finishedJob == another.finishedJob &&
- cancelledJob == another.cancelledJob &&
- failedJob == another.failedJob
+ groupId == another.groupId &&
+ totalJmMemory == another.totalTmMemory &&
+ totalTmMemory == another.totalTmMemory &&
+ totalTm == another.totalTm &&
+ totalSlot == another.totalSlot &&
+ availableSlot == another.availableSlot &&
+ runningJob == another.runningJob &&
+ finishedJob == another.finishedJob &&
+ cancelledJob == another.cancelledJob &&
+ failedJob == another.failedJob
}
}
object FlinkMetricCV {
- def empty: FlinkMetricCV = FlinkMetricCV(pollAckTime = System.currentTimeMillis)
+ def empty(groupId: String = null): FlinkMetricCV = FlinkMetricCV(groupId = groupId, pollAckTime = System.currentTimeMillis)
}
diff --git a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/model/TrackId.scala b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/model/TrackId.scala
index 7cd2d3d73..a2e3e37d9 100644
--- a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/model/TrackId.scala
+++ b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/model/TrackId.scala
@@ -25,7 +25,12 @@ import org.apache.streampark.flink.kubernetes.enums.FlinkK8sExecuteMode
/**
* tracking identifier for flink on kubernetes
*/
-case class TrackId(executeMode: FlinkK8sExecuteMode.Value, namespace: String = "default", clusterId: String, appId: Long, jobId: String) {
+case class TrackId(executeMode: FlinkK8sExecuteMode.Value,
+ namespace: String = "default",
+ clusterId: String,
+ appId: Long,
+ jobId: String,
+ groupId: String) {
def isLegal: Boolean = {
executeMode match {
@@ -44,14 +49,8 @@ case class TrackId(executeMode: FlinkK8sExecuteMode.Value, namespace: String = "
*/
def toClusterKey: ClusterKey = ClusterKey(executeMode, namespace, clusterId)
- /**
- * belong to cluster
- */
- def belongTo(clusterKey: ClusterKey): Boolean =
- executeMode == clusterKey.executeMode && namespace == clusterKey.namespace && clusterId == clusterKey.clusterId
-
override def hashCode(): Int = {
- Utils.hashCode(executeMode, clusterId, namespace, appId, jobId)
+ Utils.hashCode(executeMode, clusterId, namespace, appId, jobId, groupId)
}
override def equals(obj: Any): Boolean = {
@@ -61,7 +60,8 @@ case class TrackId(executeMode: FlinkK8sExecuteMode.Value, namespace: String = "
this.clusterId == that.clusterId &&
this.namespace == that.namespace &&
this.appId == that.appId &&
- this.jobId == that.jobId
+ this.jobId == that.jobId &&
+ this.groupId == that.groupId
case _ => false
}
}
@@ -69,11 +69,11 @@ case class TrackId(executeMode: FlinkK8sExecuteMode.Value, namespace: String = "
}
object TrackId {
- def onSession(namespace: String, clusterId: String, appId: Long, jobId: String): TrackId = {
- this(FlinkK8sExecuteMode.SESSION, namespace, clusterId, appId, jobId)
+ def onSession(namespace: String, clusterId: String, appId: Long, jobId: String, groupId: String): TrackId = {
+ this(FlinkK8sExecuteMode.SESSION, namespace, clusterId, appId, jobId, groupId)
}
- def onApplication(namespace: String, clusterId: String, appId: Long, jobId: String = null): TrackId = {
- this(FlinkK8sExecuteMode.APPLICATION, namespace, clusterId, appId, jobId)
+ def onApplication(namespace: String, clusterId: String, appId: Long, jobId: String = null, groupId: String): TrackId = {
+ this(FlinkK8sExecuteMode.APPLICATION, namespace, clusterId, appId, jobId, groupId)
}
}
diff --git a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala
index bc204959a..53c7d9f61 100644
--- a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala
+++ b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala
@@ -158,8 +158,8 @@ class FlinkJobStatusWatcher(conf: JobStatusWatcherConfig = JobStatusWatcherConfi
val appId = trackId.appId
val jobId = trackId.jobId
- val rsMap = touchSessionAllJob(clusterId, namespace, appId).toMap
- val id = TrackId.onSession(namespace, clusterId, appId, jobId)
+ val rsMap = touchSessionAllJob(clusterId, namespace, appId, trackId.groupId).toMap
+ val id = TrackId.onSession(namespace, clusterId, appId, jobId, trackId.groupId)
val jobState = rsMap.get(id).filter(_.jobState != FlinkJobState.SILENT).getOrElse {
val preCache = watchController.jobStatuses.get(id)
val state = inferSilentOrLostFromPreCache(preCache)
@@ -182,7 +182,11 @@ class FlinkJobStatusWatcher(conf: JobStatusWatcherConfig = JobStatusWatcherConfi
* This method can be called directly from outside, without affecting the
* current cachePool result.
*/
- protected[kubernetes] def touchSessionAllJob(@Nonnull clusterId: String, @Nonnull namespace: String, @Nonnull appId: Long): Array[(TrackId, JobStatusCV)] = {
+ protected[kubernetes] def touchSessionAllJob(@Nonnull clusterId: String,
+ @Nonnull namespace: String,
+ @Nonnull appId: Long,
+ @Nonnull groupId: String
+ ): Array[(TrackId, JobStatusCV)] = {
lazy val defaultResult = Array.empty[(TrackId, JobStatusCV)]
val pollEmitTime = System.currentTimeMillis
val jobDetails = listJobsDetails(ClusterKey(SESSION, namespace, clusterId)).getOrElse(return defaultResult).jobs
@@ -190,7 +194,7 @@ class FlinkJobStatusWatcher(conf: JobStatusWatcherConfig = JobStatusWatcherConfi
defaultResult
} else {
jobDetails.map { d =>
- TrackId.onSession(namespace, clusterId, appId, d.jid) -> d.toJobStatusCV(pollEmitTime, System.currentTimeMillis)
+ TrackId.onSession(namespace, clusterId, appId, d.jid, groupId) -> d.toJobStatusCV(pollEmitTime, System.currentTimeMillis)
}
}
}
diff --git a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkMetricsWatcher.scala b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkMetricsWatcher.scala
index 8fbc2b7f9..49de91848 100644
--- a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkMetricsWatcher.scala
+++ b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkMetricsWatcher.scala
@@ -144,6 +144,7 @@ class FlinkMetricWatcher(conf: MetricWatcherConfig = MetricWatcherConfig.default
val tmMemStr = flinkJmConfigs.getOrElse(TaskManagerOptions.TOTAL_PROCESS_MEMORY.key, "0b")
val jmMemStr = flinkJmConfigs.getOrElse(JobManagerOptions.TOTAL_PROCESS_MEMORY.key, "0b")
FlinkMetricCV(
+ groupId = id.groupId,
totalJmMemory = MemorySize.parse(jmMemStr).getMebiBytes,
totalTmMemory = MemorySize.parse(tmMemStr).getMebiBytes * flinkOverview.taskManagers,
totalTm = flinkOverview.taskManagers,