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 03:02:54 UTC
[incubator-streampark] 01/01: [Bug] After switch groups, the metrics dashboard info is not switched correctly. #2315
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch k8s-status
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
commit ad2111db6d016fa85b4045c03b2e424265a52c56
Author: benjobs <be...@apache.org>
AuthorDate: Tue Feb 28 11:02:34 2023 +0800
[Bug] After switch groups, the metrics dashboard info is not switched correctly. #2315
---
.../core/service/impl/ApplicationServiceImpl.java | 27 +++++++++++---------
.../console/core/task/FlinkK8sWatcherWrapper.java | 12 +++++++--
.../flink/kubernetes/DefaultFlinkK8sWatcher.scala | 5 ++--
.../flink/kubernetes/FlinkK8sWatchController.scala | 13 +++-------
.../flink/kubernetes/FlinkK8sWatcher.scala | 2 +-
.../kubernetes/FlinkK8sWatcherLazyStartAop.scala | 4 +--
.../helper/KubernetesWatcherHelper.scala | 2 +-
.../flink/kubernetes/model/FlinkMetricCV.scala | 29 ++++++++++++++--------
.../flink/kubernetes/model/TrackId.scala | 21 ++++++++--------
.../kubernetes/watcher/FlinkJobStatusWatcher.scala | 12 ++++++---
.../kubernetes/watcher/FlinkMetricsWatcher.scala | 1 +
11 files changed, 72 insertions(+), 56 deletions(-)
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 e1125ec1c..69379e874 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
@@ -268,19 +268,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);
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-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..5a2b9380a 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._
@@ -87,18 +84,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
getActiveWatchingIds() match {
case k if k.isEmpty => FlinkMetricCV.empty
case k =>
- flinkMetrics.getAll(for (elem <- k) yield {
- ClusterKey.of(elem)
- }) match {
+ 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)
+ m.values.filter(x => groupId == null || groupId == x.groupId).fold(FlinkMetricCV.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..eb8859b99 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,6 +21,7 @@ package org.apache.streampark.flink.kubernetes.model
* flink cluster metric info
*/
case class FlinkMetricCV(
+ groupId: String = null,
totalJmMemory: Integer = 0,
totalTmMemory: Integer = 0,
totalTm: Integer = 0,
@@ -33,22 +34,28 @@ case class FlinkMetricCV(
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 != another.groupId) this else {
+ 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))
+ }
+ }
}
def totalJob(): Integer = runningJob + finishedJob + cancelledJob + failedJob
def equalsPayload(another: FlinkMetricCV): Boolean = {
+ groupId == another.groupId &&
totalJmMemory == another.totalTmMemory &&
totalTmMemory == another.totalTmMemory &&
totalTm == another.totalTm &&
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..d14b091f6 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,12 +49,6 @@ 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)
}
@@ -69,11 +68,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,